<commit_before>/*
 * Copyright (c) 2010-2011 by <NAME>,
 *               Zuse Institute Berlin
 *
 * Licensed under the BSD License, see LICENSE file for details.
 *
 */

package org.xtreemfs.osd.rwre;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;

import org.xtreemfs.common.uuids.ServiceUUID;
import org.xtreemfs.common.uuids.UnknownUUIDException;
import org.xtreemfs.common.xloc.XLocations;
import org.xtreemfs.foundation.SSLOptions;
import org.xtreemfs.foundation.buffer.ASCIIString;
import org.xtreemfs.foundation.buffer.BufferPool;
import org.xtreemfs.foundation.buffer.ReusableBuffer;
import org.xtreemfs.foundation.flease.Flease;
import org.xtreemfs.foundation.flease.FleaseConfig;
import org.xtreemfs.foundation.flease.FleaseMessageSenderInterface;
import org.xtreemfs.foundation.flease.FleaseStage;
import org.xtreemfs.foundation.flease.FleaseStatusListener;
import org.xtreemfs.foundation.flease.FleaseViewChangeListenerInterface;
import org.xtreemfs.foundation.flease.comm.FleaseMessage;
import org.xtreemfs.foundation.flease.proposer.FleaseException;
import org.xtreemfs.foundation.logging.Logging;
import org.xtreemfs.foundation.logging.Logging.Category;
import org.xtreemfs.foundation.pbrpc.client.PBRPCException;
import org.xtreemfs.foundation.pbrpc.client.RPCAuthentication;
import org.xtreemfs.foundation.pbrpc.client.RPCNIOSocketClient;
import org.xtreemfs.foundation.pbrpc.client.RPCResponse;
import org.xtreemfs.foundation.pbrpc.client.RPCResponseAvailableListener;
import org.xtreemfs.foundation.pbrpc.generatedinterfaces.RPC.ErrorType;
import org.xtreemfs.foundation.pbrpc.generatedinterfaces.RPC.POSIXErrno;
import org.xtreemfs.foundation.pbrpc.generatedinterfaces.RPC.RPCHeader.ErrorResponse;
import org.xtreemfs.foundation.pbrpc.utils.ErrorUtils;
import org.xtreemfs.osd.InternalObjectData;
import org.xtreemfs.osd.OSDRequest;
import org.xtreemfs.osd.OSDRequestDispatcher;
import org.xtreemfs.osd.operations.EventRWRStatus;
import org.xtreemfs.osd.operations.OSDOperation;
import org.xtreemfs.osd.rwre.ReplicatedFileState.ReplicaState;
import org.xtreemfs.osd.stages.Stage;
import org.xtreemfs.osd.stages.StorageStage.DeleteObjectsCallback;
import org.xtreemfs.osd.stages.StorageStage.InternalGetMaxObjectNoCallback;
import org.xtreemfs.osd.stages.StorageStage.WriteObjectCallback;
import org.xtreemfs.osd.storage.CowPolicy;
import org.xtreemfs.pbrpc.generatedinterfaces.GlobalTypes.FileCredentials;
import org.xtreemfs.pbrpc.generatedinterfaces.GlobalTypes.OSDWriteResponse;
import org.xtreemfs.pbrpc.generatedinterfaces.OSD.AuthoritativeReplicaState;
import org.xtreemfs.pbrpc.generatedinterfaces.OSD.ObjectData;
import org.xtreemfs.pbrpc.generatedinterfaces.OSD.ObjectVersion;
import org.xtreemfs.pbrpc.generatedinterfaces.OSD.ObjectVersionMapping;
import org.xtreemfs.pbrpc.generatedinterfaces.OSD.ReplicaStatus;
import org.xtreemfs.pbrpc.generatedinterfaces.OSDServiceClient;

/**
 *
 * @author bjko
 */
public class RWReplicationStage extends Stage implements FleaseMessageSenderInterface {

    public static final int STAGEOP_REPLICATED_WRITE = 1;
    public static final int STAGEOP_CLOSE = 2;
    public static final int STAGEOP_PROCESS_FLEASE_MSG = 3;
    public static final int STAGEOP_PREPAREOP = 5;
    public static final int STAGEOP_TRUNCATE = 6;
    public static final int STAGEOP_GETSTATUS = 7;

    public static final int STAGEOP_INTERNAL_AUTHSTATE = 10;
    public static final int STAGEOP_INTERNAL_OBJFETCHED = 11;

    public static final int STAGEOP_LEASE_STATE_CHANGED = 13;
    public static final int STAGEOP_INTERNAL_STATEAVAIL = 14;
    public static final int STAGEOP_INTERNAL_DELETE_COMPLETE = 15;
    public static final int STAGEOP_FORCE_RESET = 16;
    public static final int STAGEOP_INTERNAL_MAXOBJ_AVAIL = 17;
    public static final int STAGEOP_INTERNAL_BACKUP_AUTHSTATE = 18;

    public  static enum Operation {
        READ,
        WRITE,
        TRUNCATE,
        INTERNAL_UPDATE,
        INTERNAL_TRUNCATE
    };


    private final RPCNIOSocketClient client;

    private final OSDServiceClient   osdClient;

    private final Map<String,ReplicatedFileState> files;

    private final Map<ASCIIString,String> cellToFileId;

    private final OSDRequestDispatcher master;

    private final FleaseStage          fstage;

    private final RPCNIOSocketClient   fleaseClient;

    private final OSDServiceClient     fleaseOsdClient;

    private final ASCIIString          localID;

    private int                        numObjsInFlight;

    private static final int           MAX_OBJS_IN_FLIGHT = 10;

    private static final int MAX_PENDING_PER_FILE = 10;

    private static final int MAX_EXTERNAL_REQUESTS_IN_Q = 250;

    private final Queue<ReplicatedFileState> filesInReset;

    private final FleaseMasterEpochThread masterEpochThread;

    private final AtomicInteger externalRequestsInQueue;



    public RWReplicationStage(OSDRequestDispatcher master, SSLOptions sslOpts) throws IOException {
        super("RWReplSt");
        this.master = master;
        client = new RPCNIOSocketClient(sslOpts, 15000, 60000*5, "RWReplicationStage");
        fleaseClient = new RPCNIOSocketClient(sslOpts, 15000, 60000*5, "RWReplicationStage (flease)");
        osdClient = new OSDServiceClient(client,null);
        fleaseOsdClient = new OSDServiceClient(fleaseClient,null);
        files = new HashMap<String, ReplicatedFileState>();
        cellToFileId = new HashMap<ASCIIString,String>();
        numObjsInFlight = 0;
        filesInReset = new LinkedList();
        externalRequestsInQueue = new AtomicInteger(0);

        localID = new ASCIIString(master.getConfig().getUUID().toString());

        masterEpochThread = new FleaseMasterEpochThread(master.getStorageStage().getStorageLayout());

        FleaseConfig fcfg = new FleaseConfig(master.getConfig().getFleaseLeaseToMS(),
                master.getConfig().getFleaseDmaxMS(), master.getConfig().getFleaseDmaxMS(),
                null, localID.toString(), master.getConfig().getFleaseRetries());

        fstage = new FleaseStage(fcfg, master.getConfig().getObjDir()+"/",
                this, false, new FleaseViewChangeListenerInterface() {

            @Override
            public void viewIdChangeEvent(ASCIIString cellId, int viewId) {
                throw new UnsupportedOperationException("Not supported yet.");
            }
        }, new FleaseStatusListener() {

            @Override
            public void statusChanged(ASCIIString cellId, Flease lease) {
                //FIXME: change state
                eventLeaseStateChanged(cellId, lease, null);
            }

            @Override
            public void leaseFailed(ASCIIString cellID, FleaseException error) {
                //change state
                //flush pending requests
                eventLeaseStateChanged(cellID, null, error);
            }
        }, masterEpochThread);
    }

    @Override
    public void start() {
        masterEpochThread.start();
        client.start();
        fleaseClient.start();
        fstage.start();
        super.start();
    }

    @Override
    public void shutdown() {
        client.shutdown();
        fleaseClient.shutdown();
        fstage.shutdown();
        masterEpochThread.shutdown();
        super.shutdown();
    }

    @Override
    public void waitForStartup() throws Exception {
        masterEpochThread.waitForStartup();
        client.waitForStartup();
        fleaseClient.waitForStartup();
        fstage.waitForStartup();
        super.waitForStartup();
    }
<commit_msg>servers: added OSDrequestDispatcher as listener to flease to ensure proper shutdown when flease stage crashes.<commit_after>/*
 * Copyright (c) 2010-2011 by <NAME>,
 *               Zuse Institute Berlin
 *
 * Licensed under the BSD License, see LICENSE file for details.
 *
 */

package org.xtreemfs.osd.rwre;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;

import org.xtreemfs.common.uuids.ServiceUUID;
import org.xtreemfs.common.uuids.UnknownUUIDException;
import org.xtreemfs.common.xloc.XLocations;
import org.xtreemfs.foundation.SSLOptions;
import org.xtreemfs.foundation.buffer.ASCIIString;
import org.xtreemfs.foundation.buffer.BufferPool;
import org.xtreemfs.foundation.buffer.ReusableBuffer;
import org.xtreemfs.foundation.flease.Flease;
import org.xtreemfs.foundation.flease.FleaseConfig;
import org.xtreemfs.foundation.flease.FleaseMessageSenderInterface;
import org.xtreemfs.foundation.flease.FleaseStage;
import org.xtreemfs.foundation.flease.FleaseStatusListener;
import org.xtreemfs.foundation.flease.FleaseViewChangeListenerInterface;
import org.xtreemfs.foundation.flease.comm.FleaseMessage;
import org.xtreemfs.foundation.flease.proposer.FleaseException;
import org.xtreemfs.foundation.logging.Logging;
import org.xtreemfs.foundation.logging.Logging.Category;
import org.xtreemfs.foundation.pbrpc.client.PBRPCException;
import org.xtreemfs.foundation.pbrpc.client.RPCAuthentication;
import org.xtreemfs.foundation.pbrpc.client.RPCNIOSocketClient;
import org.xtreemfs.foundation.pbrpc.client.RPCResponse;
import org.xtreemfs.foundation.pbrpc.client.RPCResponseAvailableListener;
import org.xtreemfs.foundation.pbrpc.generatedinterfaces.RPC.ErrorType;
import org.xtreemfs.foundation.pbrpc.generatedinterfaces.RPC.POSIXErrno;
import org.xtreemfs.foundation.pbrpc.generatedinterfaces.RPC.RPCHeader.ErrorResponse;
import org.xtreemfs.foundation.pbrpc.utils.ErrorUtils;
import org.xtreemfs.osd.InternalObjectData;
import org.xtreemfs.osd.OSDRequest;
import org.xtreemfs.osd.OSDRequestDispatcher;
import org.xtreemfs.osd.operations.EventRWRStatus;
import org.xtreemfs.osd.operations.OSDOperation;
import org.xtreemfs.osd.rwre.ReplicatedFileState.ReplicaState;
import org.xtreemfs.osd.stages.Stage;
import org.xtreemfs.osd.stages.StorageStage.DeleteObjectsCallback;
import org.xtreemfs.osd.stages.StorageStage.InternalGetMaxObjectNoCallback;
import org.xtreemfs.osd.stages.StorageStage.WriteObjectCallback;
import org.xtreemfs.osd.storage.CowPolicy;
import org.xtreemfs.pbrpc.generatedinterfaces.GlobalTypes.FileCredentials;
import org.xtreemfs.pbrpc.generatedinterfaces.GlobalTypes.OSDWriteResponse;
import org.xtreemfs.pbrpc.generatedinterfaces.OSD.AuthoritativeReplicaState;
import org.xtreemfs.pbrpc.generatedinterfaces.OSD.ObjectData;
import org.xtreemfs.pbrpc.generatedinterfaces.OSD.ObjectVersion;
import org.xtreemfs.pbrpc.generatedinterfaces.OSD.ObjectVersionMapping;
import org.xtreemfs.pbrpc.generatedinterfaces.OSD.ReplicaStatus;
import org.xtreemfs.pbrpc.generatedinterfaces.OSDServiceClient;

/**
 *
 * @author bjko
 */
public class RWReplicationStage extends Stage implements FleaseMessageSenderInterface {

    public static final int STAGEOP_REPLICATED_WRITE = 1;
    public static final int STAGEOP_CLOSE = 2;
    public static final int STAGEOP_PROCESS_FLEASE_MSG = 3;
    public static final int STAGEOP_PREPAREOP = 5;
    public static final int STAGEOP_TRUNCATE = 6;
    public static final int STAGEOP_GETSTATUS = 7;

    public static final int STAGEOP_INTERNAL_AUTHSTATE = 10;
    public static final int STAGEOP_INTERNAL_OBJFETCHED = 11;

    public static final int STAGEOP_LEASE_STATE_CHANGED = 13;
    public static final int STAGEOP_INTERNAL_STATEAVAIL = 14;
    public static final int STAGEOP_INTERNAL_DELETE_COMPLETE = 15;
    public static final int STAGEOP_FORCE_RESET = 16;
    public static final int STAGEOP_INTERNAL_MAXOBJ_AVAIL = 17;
    public static final int STAGEOP_INTERNAL_BACKUP_AUTHSTATE = 18;

    public  static enum Operation {
        READ,
        WRITE,
        TRUNCATE,
        INTERNAL_UPDATE,
        INTERNAL_TRUNCATE
    };


    private final RPCNIOSocketClient client;

    private final OSDServiceClient   osdClient;

    private final Map<String,ReplicatedFileState> files;

    private final Map<ASCIIString,String> cellToFileId;

    private final OSDRequestDispatcher master;

    private final FleaseStage          fstage;

    private final RPCNIOSocketClient   fleaseClient;

    private final OSDServiceClient     fleaseOsdClient;

    private final ASCIIString          localID;

    private int                        numObjsInFlight;

    private static final int           MAX_OBJS_IN_FLIGHT = 10;

    private static final int MAX_PENDING_PER_FILE = 10;

    private static final int MAX_EXTERNAL_REQUESTS_IN_Q = 250;

    private final Queue<ReplicatedFileState> filesInReset;

    private final FleaseMasterEpochThread masterEpochThread;

    private final AtomicInteger externalRequestsInQueue;



    public RWReplicationStage(OSDRequestDispatcher master, SSLOptions sslOpts) throws IOException {
        super("RWReplSt");
        this.master = master;
        client = new RPCNIOSocketClient(sslOpts, 15000, 60000*5, "RWReplicationStage");
        fleaseClient = new RPCNIOSocketClient(sslOpts, 15000, 60000*5, "RWReplicationStage (flease)");
        osdClient = new OSDServiceClient(client,null);
        fleaseOsdClient = new OSDServiceClient(fleaseClient,null);
        files = new HashMap<String, ReplicatedFileState>();
        cellToFileId = new HashMap<ASCIIString,String>();
        numObjsInFlight = 0;
        filesInReset = new LinkedList();
        externalRequestsInQueue = new AtomicInteger(0);

        localID = new ASCIIString(master.getConfig().getUUID().toString());

        masterEpochThread = new FleaseMasterEpochThread(master.getStorageStage().getStorageLayout());

        FleaseConfig fcfg = new FleaseConfig(master.getConfig().getFleaseLeaseToMS(),
                master.getConfig().getFleaseDmaxMS(), master.getConfig().getFleaseDmaxMS(),
                null, localID.toString(), master.getConfig().getFleaseRetries());

        fstage = new FleaseStage(fcfg, master.getConfig().getObjDir()+"/",
                this, false, new FleaseViewChangeListenerInterface() {

            @Override
            public void viewIdChangeEvent(ASCIIString cellId, int viewId) {
                throw new UnsupportedOperationException("Not supported yet.");
            }
        }, new FleaseStatusListener() {

            @Override
            public void statusChanged(ASCIIString cellId, Flease lease) {
                //FIXME: change state
                eventLeaseStateChanged(cellId, lease, null);
            }

            @Override
            public void leaseFailed(ASCIIString cellID, FleaseException error) {
                //change state
                //flush pending requests
                eventLeaseStateChanged(cellID, null, error);
            }
        }, masterEpochThread);
        fstage.setLifeCycleListener(master);
    }

    @Override
    public void start() {
        masterEpochThread.start();
        client.start();
        fleaseClient.start();
        fstage.start();
        super.start();
    }

    @Override
    public void shutdown() {
        client.shutdown();
        fleaseClient.shutdown();
        fstage.shutdown();
        masterEpochThread.shutdown();
        super.shutdown();
    }

    @Override
    public void waitForStartup() throws Exception {
        masterEpochThread.waitForStartup();
        client.waitForStartup();
        fleaseClient.waitForStartup();
        fstage.waitForStartup();
        super.waitForStartup();
    }

{'fraction_non_alphanumeric': 0.08240988409119858, 'fraction_numerical': 0.0077697108648579795, 'mean_word_length': 3.8257529194837123, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2559729', 'n_tokens_mistral': 4959, 'n_tokens_neox': 4758, 'n_words': 893}