<commit_before>#include <assert.h>
#include <limits.h>
#include <signal.h>
#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <unistd.h>

#include "adios2/ADIOSConfig.h"
#include <atl.h>
#include <evpath.h>
#ifdef ADIOS2_HAVE_MPI
#include <mpi.h>
#else
#include "sstmpidummy.h"
#endif
#include <pthread.h>

#include "sst.h"

#include "cp_internal.h"

extern void CP_verbose(SstStream Stream, char *Format, ...);

static void sendOneToEachWriterRank(SstStream s, CMFormat f, void *Msg,
                                    void **WS_StreamPtr);
static void CP_PeerFailCloseWSReader(WS_ReaderInfo CP_WSR_Stream,
                                     enum StreamStatus NewState);

#ifdef MUTEX_DEBUG
#define PTHREAD_MUTEX_LOCK(lock)                                               \
    printf("Trying lock line %d\n", __LINE__);                                 \
    pthread_mutex_lock(lock);                                                  \
    printf("Got lock\n");
#define PTHREAD_MUTEX_UNLOCK(lock)                                             \
    printf("UNlocking line %d\n", __LINE__);                                   \
    pthread_mutex_unlock(lock);
#else
#define PTHREAD_MUTEX_LOCK(lock) pthread_mutex_lock(lock);
#define PTHREAD_MUTEX_UNLOCK(lock) pthread_mutex_unlock(lock);
#endif

static char *buildContactInfo(SstStream Stream)
{
    char *Contact = CP_GetContactString(Stream);
    char *FullInfo = malloc(strlen(Contact) + 20);
    sprintf(FullInfo, "%p:%s", (void *)Stream, Contact);
    free(Contact);
    return FullInfo;
}

struct NameListEntry
{
    const char *FileName;
    struct NameListEntry *Next;
};

struct NameListEntry *FileNameList = NULL;

static void RemoveAllFilesInList()
{
    while (FileNameList)
    {
        struct NameListEntry *Next = FileNameList->Next;
        fprintf(stderr, "SST stream open at exit, unlinking contact file %s\n",
                FileNameList->FileName);
        unlink(FileNameList->FileName);
        free(FileNameList);
        FileNameList = Next;
    }
}

static void ExitAndRemoveFiles(int signum)
{
    fprintf(stderr, "ADIOS2 caught SigInt, exiting with error\n");
    exit(1);
}

static void AddNameToExitList(const char *FileName)
{
    static int First = 1;
    if (First)
    {
        struct sigaction sa;
        First = 0;
        atexit(RemoveAllFilesInList);

        memset(&sa, 0, sizeof(sa));
        sa.sa_handler = ExitAndRemoveFiles;
        sigemptyset(&sa.sa_mask);
        sigaction(SIGINT, &sa, NULL);
    }

    struct NameListEntry *NewHead = malloc(sizeof(*NewHead));
    NewHead->FileName = FileName;
    NewHead->Next = FileNameList;
    FileNameList = NewHead;
}

static void RemoveNameFromExitList(const char *FileName)
{
    struct NameListEntry **LastPtr = &FileNameList;
    struct NameListEntry *List = FileNameList;
    while (List)
    {
        if (strcmp(FileName, List->FileName) == 0)
        {
            *LastPtr = List->Next;
            free(List);
            return;
        }
        List = List->Next;
    }
}

static void writeContactInfoFile(const char *Name, SstStream Stream)
{
    char *Contact = buildContactInfo(Stream);
    char *TmpName = malloc(strlen(Name) + strlen(".tmp") + 1);
    char *FileName = malloc(strlen(Name) + strlen(SST_POSTFIX) + 1);
    FILE *WriterInfo;

    /*
     * write the contact information file with a temporary name before
     * renaming it to the final version to help prevent partial reads
     */
    sprintf(TmpName, "%s.tmp", Name);
    sprintf(FileName, "%s" SST_POSTFIX, Name);
    WriterInfo = fopen(TmpName, "w");
    fprintf(WriterInfo, "%s", Contact);
    fclose(WriterInfo);
    rename(TmpName, FileName);
    Stream->AbsoluteFilename = realpath(FileName, NULL);
    free(Contact);
    free(TmpName);
    free(FileName);
    AddNameToExitList(Stream->AbsoluteFilename);
}

static void writeContactInfoScreen(const char *Name, SstStream Stream)
{
    char *Contact = buildContactInfo(Stream);

    /*
     * write the contact information file to the screen
     */
    fprintf(stdout, "The next line of output is the contact information "
                    "associated with SST output stream \"%s\".  Please make it "
                    "available to the reader.\n",
            Name);
    fprintf(stdout, "\t%s\n", Contact);
    free(Contact);
}

static void registerContactInfo(const char *Name, SstStream Stream)
{
    switch (Stream->RegistrationMethod)
    {
    case SstRegisterFile:
        writeContactInfoFile(Name, Stream);
        break;
    case SstRegisterScreen:
        writeContactInfoScreen(Name, Stream);
        break;
    case SstRegisterCloud:
        /* not yet */
        break;
    }
}

static void removeContactInfoFile(SstStream Stream)
{
    const char *Name = Stream->AbsoluteFilename;
    unlink(Name);
    RemoveNameFromExitList(Name);
}

static void removeContactInfo(SstStream Stream)
{
    switch (Stream->RegistrationMethod)
    {
    case SstRegisterFile:
        removeContactInfoFile(Stream);
        break;
    case SstRegisterScreen:
        /* nothing necessary here */
        break;
    case SstRegisterCloud:
        /* not yet */
        break;
    }
}

static void WriterConnCloseHandler(CManager cm, CMConnection closed_conn,
                                   void *client_data)
{
    WS_ReaderInfo WSreader = (WS_ReaderInfo)client_data;
    SstStream ParentWriterStream = WSreader->ParentStream;

    PTHREAD_MUTEX_LOCK(&ParentWriterStream->DataLock);
    if (WSreader->ReaderStatus == Established)
    {
        /*
         * tag our reader instance as failed.
         * If any instance is failed, we should remove all, but that requires a
         * global operation, so prep.
         */
        CP_verbose(ParentWriterStream, "Writer-side Rank received a "
                                       "connection-close event during normal "
                                       "operations, peer likely failed\n");
        PTHREAD_MUTEX_UNLOCK(&ParentWriterStream->DataLock);
        CP_PeerFailCloseWSReader(WSreader, PeerFailed);
        PTHREAD_MUTEX_LOCK(&ParentWriterStream->DataLock);
        ParentWriterStream->GlobalOpRequired = 1;
    }
    else if ((WSreader->ReaderStatus == PeerClosed) ||
             (WSreader->ReaderStatus == Closed))
    {
        /* ignore this.  We expect a close after the connection is marked closed
         */
        CP_verbose(ParentWriterStream, "Writer-side Rank received a "
                                       "connection-close event after close, "
                                       "not unexpected\n");
    }
    else
    {
        fprintf(stderr, "Got an unexpected connection close event\n");
        CP_verbose(ParentWriterStream, "Writer-side Rank received a "
                                       "connection-close event in unexpected "
                                       "state %s\n",
                   SSTStreamStatusStr[WSreader->ReaderStatus]);
        PTHREAD_MUTEX_UNLOCK(&ParentWriterStream->DataLock);
        CP_PeerFailCloseWSReader(WSreader, PeerFailed);
        PTHREAD_MUTEX_LOCK(&ParentWriterStream->DataLock);
    }
    /* main thread might be waiting on status change */
    pthread_cond_signal(&ParentWriterStream->DataCondition);
    PTHREAD_MUTEX_UNLOCK(&ParentWriterStream->DataLock);
}

static void SendPeerSetupMsg(WS_ReaderInfo reader, int reversePeer, int myRank)
{
    CMConnection conn = reader->Connections[reversePeer].CMconn;
    SstStream Stream = reader->ParentStream;
    struct _PeerSetupMsg setup;
    setup.RS_Stream = reader->Connections[reversePeer].RemoteStreamID;
    setup.WriterRank = myRank;
    setup.WriterCohortSize = Stream->CohortSize;
    if (CMwrite(conn, Stream->CPInfo->PeerSetupFormat, &setup) != 1)
    {
        CP_verbose(Stream,
                   "Message failed to send to reader in sendPeerSetup in "
                   "reader open\n");
    }
}

static int initWSReader(WS_ReaderInfo reader, int ReaderSize,
                        CP_ReaderInitInfo *reader_info)
{
    int WriterSize = reader->ParentStream->CohortSize;
    int WriterRank = reader->ParentStream->Rank;
    int i;
    int *reverseArray;
    reader->ReaderCohortSize = ReaderSize;
    if (!reader->Connections)
    {
        reader->Connections =
            calloc(sizeof(reader->Connections[0]), ReaderSize);
    }
    for (i = 0; i < ReaderSize; i++)
    {
        if (!reader->Connections[i].ContactList)
        {
            reader->Connections[i].ContactList =
                attr_list_from_string(reader_info[i]->ContactInfo);
        }
        reader->Connections[i].RemoteStreamID = reader_info[i]->ReaderID;
        reader->Connections[i].CMconn = NULL;
    }
    /*
     *   Peering.
     *   We use peering for two things:
     *     - failure awareness (each rank needs a close handler on one
     connection to some opposite rank so they can detect failure)
     *     - notification (how info gets sent from reader to writer and vice
     versa)
     *
     *   A connection that exists for notification is also useful for
     *   failure awareness, but where not are necessary for
     *   notification, we still may make some for failure
     *   notification.

     *   In this code, all connections are made by the writing side,
     *   but the reader side must be sent notifications so that it is
     *   aware of what connections are made for it and what they are
     *   to be used for (I.E. notification, or only existing passively
     *   for failure notification).
     *
     *   Connections that are used for notification from writer to
     *   reader will be in the Peer list and we'll send messages down
     *   them later.  If there are many more writers than readers
     *   (presumed normal case), the peer list will have 0 or 1
     *   entries.  Connections in the reverseArray are for failure
     *   awareness and/or notification from reader to writer.  If
     *   there are many more readers than writers, the reverseArray
     *   will have one entry (to the one reader that will send us
     *   notifications and which we will use for failure awareness).

     *   If there are equal numbers of readers and writers, then each
     *   rank is peered only with the same rank in the opposing.

     *   If there happen to be many more readers than writers, then
     *   the Peer list will contain a lot of entries (all those that
     *   get notifications from us.  The reverseArray will also
     *   contain a lot of entries, but only the first will send us
     *   notifications.  The others will just use the connections for
     *   failure awareness.

     */
    getPeerArrays(WriterSize, WriterRank, ReaderSize, &reader->Peers,
                  &reverseArray);

    i = 0;
    while (reverseArray[i] != -1)
    {
        int peer = reverseArray[i];
        if (reader->ParentStream->ConnectionUsleepMultiplier != 0)
            usleep(WriterRank *
                   reader->ParentStream->ConnectionUsleepMultiplier);
        reader->Connections[peer].CMconn =
            CMget_conn(reader->ParentStream->CPInfo->cm,
                       reader->Connections[peer].ContactList);

        if (!reader->Connections[peer].CMconn)
        {
            CP_error(
                reader->ParentStream,
                "Connection failed in SstInitWSReader! Contact list was:\n");
            CP_error(
                reader->ParentStream, "%s\n",
                attr_list_to_string(reader->Connections[peer].ContactList));
            /* fail the stream */
            return 0;
        }

        CMconn_register_close_handler(reader->Connections[peer].CMconn,
                                      WriterConnCloseHandler, (void *)reader);
        if (i == 0)
        {
            /* failure awareness for reader rank */
            CP_verbose(reader->ParentStream, "Sending peer setup to rank %d\n",
                       peer);
            SendPeerSetupMsg(reader, peer, reader->ParentStream->Rank);
        }
        else
        {
            CP_verbose(reader->ParentStream, "Sending peer setup to rank %d\n",
                       peer);
            /* failure awareness for reader rank */
            SendPeerSetupMsg(reader, peer, -1);
        }
        i++;
    }
    free(reverseArray);
    i = 0;
    while (reader->Peers[i] != -1)
    {
        int peer = reader->Peers[i];
        if (reader->Connections[peer].CMconn)
        {
            /* already made this above */
            i++;
            continue;
        }
        if (reader->ParentStream->ConnectionUsleepMultiplier != 0)
            usleep(WriterRank *
                   reader->ParentStream->ConnectionUsleepMultiplier);
        reader->Connections[peer].CMconn =
            CMget_conn(reader->ParentStream->CPInfo->cm,
                       reader->Connections[peer].ContactList);

        if (!reader->Connections[peer].CMconn)
        {
            CP_error(
                reader->ParentStream,
                "Connection failed in SstInitWSReader! Contact list was:\n");
            CP_error(
                reader->ParentStream, "%s\n",
                attr_list_to_string(reader->Connections[peer].ContactList));
            /* fail the stream */
            return 0;
        }

        CMconn_register_close_handler(reader->Connections[peer].CMconn,
                                      WriterConnCloseHandler, (void *)reader);
        /* failure awareness for reader rank */
        CP_verbose(reader->ParentStream, "Sending peer setup to rank %d\n",
                   peer);
        SendPeerSetupMsg(reader, peer, reader->ParentStream->Rank);
        i++;
    }
    return 1;
}

static long earliestAvailableTimestepNumber(SstStream Stream,
                                            long CurrentTimestep)
{
    long Ret = CurrentTimestep;
    CPTimestepList List = Stream->QueuedTimesteps;
    PTHREAD_MUTEX_LOCK(&Stream->DataLock);
    while (List)
    {
        if (List->Timestep < Ret)
        {
            List->ReferenceCount++;
            Ret = List->Timestep;
            CP_verbose(Stream, "Earliest available : Writer-side Timestep %ld "
                               "now has reference count %d\n",
                       List->Timestep, List->ReferenceCount);
        }
        List = List->Next;
    }
    PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
    return Ret;
}

static void SubRefRangeTimestep(SstStream Stream, long LowRange, long HighRange)
{
    CPTimestepList Last = NULL, List;
    int AnythingRemoved = 0;
    List = Stream->QueuedTimesteps;
    while (List)
    {
        if ((List->Timestep >= LowRange) && (List->Timestep <= HighRange))
        {
            List->ReferenceCount--;
            CP_verbose(Stream, "SubRef : Writer-side Timestep %ld now has "
                               "reference count %d\n",
                       List->Timestep, List->ReferenceCount);
        }
        if (List->ReferenceCount == 0)
        {
            /* dequeue and free */
            CPTimestepList ItemToFree = List;
            if (Last == NULL)
            {
                Stream->QueuedTimesteps = List->Next;
            }
            else
            {
                Last->Next = List->Next;
            }
            CP_verbose(Stream, "Step %d reference count reached zero, "
                               "releasing from DP and freeing\n",
                       ItemToFree->Timestep);
            Stream->DP_Interface->releaseTimestep(&Svcs, Stream->DP_Stream,
                                                  List->Timestep);

            Stream->QueuedTimestepCount--;
            List = List->Next;
            ItemToFree->FreeTimestep(ItemToFree->FreeClientData);
            free(ItemToFree->Msg);
            free(ItemToFree->MetadataArray);
            free(ItemToFree->DP_TimestepInfo);
            free(ItemToFree->DataBlockToFree);
            free(ItemToFree);
            AnythingRemoved++;
        }
        else
        {
            Last = List;
            List = List->Next;
        }
    }
    if (AnythingRemoved)
    {
        /* main thread might be waiting on timesteps going away */
        pthread_cond_signal(&Stream->DataCondition);
    }
}

static void KillZeroRefTimesteps(SstStream Stream)
{
    CPTimestepList Last = NULL, List;
    int AnythingRemoved = 0;
    PTHREAD_MUTEX_LOCK(&Stream->DataLock);
    List = Stream->QueuedTimesteps;
    while (List)
    {
        if (List->ReferenceCount == 0)
        {
            /* dequeue and free */
            CPTimestepList ItemToFree = List;
            if (Last == NULL)
            {
                Stream->QueuedTimesteps = List->Next;
            }
            else
            {
                Last->Next = List->Next;
            }
            CP_verbose(Stream, "Step %d reference count is zero, "
                               "releasing from DP and freeing\n",
                       List->Timestep);
            Stream->DP_Interface->releaseTimestep(&Svcs, Stream->DP_Stream,
                                                  List->Timestep);

            Stream->QueuedTimestepCount--;
            List = List->Next;
            free(ItemToFree);
            AnythingRemoved++;
        }
        else
        {
            Last = List;
            List = List->Next;
        }
    }
    if (AnythingRemoved)
    {
        /* main thread might be waiting on timesteps going away */
        pthread_cond_signal(&Stream->DataCondition);
    }
    PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
}

WS_ReaderInfo WriterParticipateInReaderOpen(SstStream Stream)
{
    RequestQueue Req;
    reader_data_t ReturnData;
    void *free_block = NULL;
    int WriterResponseCondition = -1;
    CMConnection conn;
    long MyStartingTimestep, GlobalStartingTimestep;

    CP_verbose(Stream, "Beginning writer-side reader open protocol\n");
    if (Stream->Rank == 0)
    {
        PTHREAD_MUTEX_LOCK(&Stream->DataLock);
        assert((Stream->ReadRequestQueue));
        Req = Stream->ReadRequestQueue;
        Stream->ReadRequestQueue = Req->Next;
        Req->Next = NULL;
        PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
        struct _CombinedReaderInfo reader_data;
        memset(&reader_data, 0, sizeof(reader_data));
        reader_data.ReaderCohortSize = Req->Msg->ReaderCohortSize;
        reader_data.CP_ReaderInfo = Req->Msg->CP_ReaderInfo;
        reader_data.DP_ReaderInfo = Req->Msg->DP_ReaderInfo;
        ReturnData = CP_distributeDataFromRankZero(
            Stream, &reader_data, Stream->CPInfo->CombinedReaderInfoFormat,
            &free_block);
        WriterResponseCondition = Req->Msg->WriterResponseCondition;
        conn = Req->Conn;
        CMreturn_buffer(Stream->CPInfo->cm, Req->Msg);
        free(Req);
    }
    else
    {
        ReturnData = CP_distributeDataFromRankZero(
            Stream, NULL, Stream->CPInfo->CombinedReaderInfoFormat,
            &free_block);
    }
    //    printf("I am writer rank %d, my info on readers is:\n", Stream->Rank);
    //    FMdump_data(FMFormat_of_original(Stream->CPInfo->combined_reader_Format),
    //                ReturnData, 1024000);
    //    printf("\n");

    Stream->Readers = realloc(Stream->Readers, sizeof(Stream->Readers[0]) *
                                                   (Stream->ReaderCount + 1));
    DP_WSR_Stream per_reader_Stream;
    void *DP_WriterInfo;
    void *ret_data_block;
    CP_PeerConnection *connections_to_reader;
    connections_to_reader =
        calloc(sizeof(CP_PeerConnection), ReturnData->ReaderCohortSize);
    for (int i = 0; i < ReturnData->ReaderCohortSize; i++)
    {
        attr_list attrs =
            attr_list_from_string(ReturnData->CP_ReaderInfo[i]->ContactInfo);
        connections_to_reader[i].ContactList = attrs;
        connections_to_reader[i].RemoteStreamID =
            ReturnData->CP_ReaderInfo[i]->ReaderID;
    }

    per_reader_Stream = Stream->DP_Interface->initWriterPerReader(
        &Svcs, Stream->DP_Stream, ReturnData->ReaderCohortSize,
        connections_to_reader, ReturnData->DP_ReaderInfo, &DP_WriterInfo);

    WS_ReaderInfo CP_WSR_Stream = malloc(sizeof(*CP_WSR_Stream));
    CP_WSR_Stream->ReaderStatus = NotOpen;
    Stream->Readers[Stream->ReaderCount] = CP_WSR_Stream;
    CP_WSR_Stream->DP_WSR_Stream = per_reader_Stream;
    CP_WSR_Stream->ParentStream = Stream;
    CP_WSR_Stream->Connections = connections_to_reader;

    int MySuccess = initWSReader(CP_WSR_Stream, ReturnData->ReaderCohortSize,
                                 ReturnData->CP_ReaderInfo);

    int GlobalSuccess = 0;
    MPI_Allreduce(&MySuccess, &GlobalSuccess, 1, MPI_INT, MPI_LAND,
                  Stream->mpiComm);

    if (!GlobalSuccess)
    {
        return NULL;
    }
    AddToLastCallFreeList(CP_WSR_Stream);
    free(free_block);
    ReturnData = NULL; /* now invalid */

    Stream->ReaderCount++;

    struct _CP_DP_PairInfo combined_init;
    struct _CP_WriterInitInfo cpInfo;

    struct _CP_DP_PairInfo **pointers = NULL;

    memset(&cpInfo, 0, sizeof(cpInfo));
    cpInfo.ContactInfo = CP_GetContactString(Stream);
    cpInfo.WriterID = CP_WSR_Stream;

    combined_init.CP_Info = (void **)&cpInfo;
    combined_init.DP_Info = DP_WriterInfo;

    MyStartingTimestep =
        earliestAvailableTimestepNumber(Stream, Stream->WriterTimestep);
    if (MyStartingTimestep == -1)
        MyStartingTimestep = 0;

    MPI_Allreduce(&MyStartingTimestep, &GlobalStartingTimestep, 1, MPI_LONG,
                  MPI_MAX, Stream->mpiComm);

    /*
     *  The earliestAvailableTimestepNumber subroutine also added to the
     *  reference count of all existing timesteps, in case we will be making
     *  it available to the joining reader.  Once we've determined what the
     *  starting timestep we'll make available is (GlobalStartingTimestep),
     *  decrement the ref count of others.
     */
    if (MyStartingTimestep != GlobalStartingTimestep)
    {
        CP_verbose(Stream, "In writer participate in reader open, releasing "
                           "timesteps from %ld to %ld\n",
                   MyStartingTimestep, GlobalStartingTimestep - 1);

        SubRefRangeTimestep(Stream, MyStartingTimestep,
                            GlobalStartingTimestep - 1);
    }
    CP_verbose(Stream,
               "My oldest timestep was %ld, global oldest timestep was %ld\n",
               MyStartingTimestep, GlobalStartingTimestep);

    CP_WSR_Stream->StartingTimestep = GlobalStartingTimestep;

    pointers = (struct _CP_DP_PairInfo **)CP_consolidateDataToRankZero(
        Stream, &combined_init, Stream->CPInfo->PerRankWriterInfoFormat,
        &ret_data_block);

    if (Stream->Rank == 0)
    {
        struct _WriterResponseMsg response;
        memset(&response, 0, sizeof(response));
        response.WriterResponseCondition = WriterResponseCondition;
        response.WriterCohortSize = Stream->CohortSize;
        response.WriterConfigParams = Stream->ConfigParams;
        response.NextStepNumber = GlobalStartingTimestep;
        response.CP_WriterInfo =
            malloc(response.WriterCohortSize * sizeof(void *));
        response.DP_WriterInfo =
            malloc(response.WriterCohortSize * sizeof(void *));
        for (int i = 0; i < response.WriterCohortSize; i++)
        {
            response.CP_WriterInfo[i] =
                (struct _CP_WriterInitInfo *)pointers[i]->CP_Info;
            response.DP_WriterInfo[i] = pointers[i]->DP_Info;
        }
        if (CMwrite(conn, Stream->CPInfo->WriterResponseFormat, &response) != 1)
        {
            CP_verbose(Stream,
                       "Message failed to send to reader in participate in "
                       "reader open\n");
        }
        free(response.CP_WriterInfo);
        free(response.DP_WriterInfo);
    }
    free(cpInfo.ContactInfo);
    if (ret_data_block)
        free(ret_data_block);
    if (pointers)
        free(pointers);
    Stream->NewReaderPresent = 1;
    CP_verbose(Stream, "Finish writer-side reader open protocol for reader %p, "
                       "reader ready response pending\n",
               CP_WSR_Stream);
    return CP_WSR_Stream;
}

void sendOneToWSRCohort(WS_ReaderInfo CP_WSR_Stream, CMFormat f, void *Msg,
                        void **RS_StreamPtr)
{
    SstStream s = CP_WSR_Stream->ParentStream;
    int j = 0;

    while (CP_WSR_Stream->Peers[j] != -1)
    {
        int peer = CP_WSR_Stream->Peers[j];
        CMConnection conn = CP_WSR_Stream->Connections[peer].CMconn;
        /* add the reader-rank-specific Stream identifier to each outgoing
         * message */
        *RS_StreamPtr = CP_WSR_Stream->Connections[peer].RemoteStreamID;
        CP_verbose(s, "Sending a message to reader %d (%p)\n", peer,
                   *RS_StreamPtr);
        if (CMwrite(conn, f, Msg) != 1)
        {
            CP_verbose(s, "Message failed to send to reader %d (%p)\n", peer,
                       *RS_StreamPtr);
        }
        j++;
    }
}

static void waitForReaderResponseAndSendQueued(WS_ReaderInfo Reader)
{
    SstStream Stream = Reader->ParentStream;
    PTHREAD_MUTEX_LOCK(&Stream->DataLock);
    while (Reader->ReaderStatus != Established)
    {
        /* NEED TO HANDLE FAILURE HERE */
        CP_verbose(Stream, "Waiting for Reader ready on WSR %p.\n", Reader);
        pthread_cond_wait(&Stream->DataCondition, &Stream->DataLock);
    }
    assert(Reader->ReaderStatus == Established);
    PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);

    /* send any queued metadata necessary */
    Reader->OldestUnreleasedTimestep = Reader->StartingTimestep;
    CP_verbose(Stream, "Reader ready on WSR %p, Stream established.\n", Reader);
    for (long TS = Reader->StartingTimestep; TS <= Stream->LastProvidedTimestep;
         TS++)
    {
        CPTimestepList List = Stream->QueuedTimesteps;
        while (List)
        {
            if (List->Timestep == TS)
            {
                FFSFormatList SavedFormats = List->Msg->Formats;
                CP_verbose(Stream,
                           "Sending Queued TimestepMetadata for timestep %d\n",
                           TS);

                if (TS == Reader->StartingTimestep)
                {
                    /* For first Msg, send all previous formats */
                    List->Msg->Formats = Stream->PreviousFormats;
                }
                else
                {
                    /*
                     *  TENTATIVE!  TRYING TO SEE IF THIS MIGHT IMPACT RARE
                     *  STUCK READER PROBLEM.
                     *  Add  a short delay between consecutive messages
                     */
                    usleep(10 * Stream->ConnectionUsleepMultiplier);
                }
                sendOneToWSRCohort(
                    Reader, Stream->CPInfo->DeliverTimestepMetadataFormat,
                    List->Msg, &List->Msg->RS_Stream);
                if (TS == Reader->StartingTimestep)
                {
                    /* restore Msg format list */
                    List->Msg->Formats = SavedFormats;
                }
            }
            List = List->Next;
        }
    }
}

SstStream SstWriterOpen(const char *Name, SstParams Params, MPI_Comm comm)
{
    SstStream Stream;

    Stream = CP_newStream();
    Stream->Role = WriterRole;
    CP_validateParams(Stream, Params, 1 /* Writer */);
    Stream->ConfigParams = Params;

    char *Filename = strdup(Name);

    Stream->mpiComm = comm;

    MPI_Comm_rank(Stream->mpiComm, &Stream->Rank);
    MPI_Comm_size(Stream->mpiComm, &Stream->CohortSize);

    Stream->DP_Interface = SelectDP(&Svcs, Stream, Stream->ConfigParams);

    if (!Stream->DP_Interface)
    {
        CP_verbose(Stream, "Failed to load DataPlane %s for Stream \"%s\"\n",
                   Params->DataTransport, Filename);
        return NULL;
    }

    Stream->CPInfo = CP_getCPInfo(Stream->DP_Interface);

    if (Stream->RendezvousReaderCount > 0)
    {
        Stream->FirstReaderCondition =
            CMCondition_get(Stream->CPInfo->cm, NULL);
    }
    else
    {
        Stream->FirstReaderCondition = -1;
    }

    Stream->DP_Stream =
        Stream->DP_Interface->initWriter(&Svcs, Stream, Stream->ConfigParams);

    if (Stream->Rank == 0)
    {
        registerContactInfo(Filename, Stream);
    }

    CP_verbose(Stream, "Opening Stream \"%s\"\n", Filename);

    if (Stream->Rank == 0)
    {
        CP_verbose(Stream, "Writer stream params are:\n");
        CP_dumpParams(Stream, Stream->ConfigParams);
    }

    while (Stream->RendezvousReaderCount > 0)
    {
        WS_ReaderInfo reader;
        CP_verbose(Stream, "Stream \"%s\" waiting for %d readers\n", Filename,
                   Stream->RendezvousReaderCount);
        if (Stream->Rank == 0)
        {
            PTHREAD_MUTEX_LOCK(&Stream->DataLock);
            if (Stream->ReadRequestQueue == NULL)
            {
                pthread_cond_wait(&Stream->DataCondition, &Stream->DataLock);
            }
            assert(Stream->ReadRequestQueue);
            PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
        }
        MPI_Barrier(Stream->mpiComm);

        struct timeval Start, Stop, Diff;
        gettimeofday(&Start, NULL);
        reader = WriterParticipateInReaderOpen(Stream);
        if (!reader)
        {
            CP_error(Stream, "Potential reader registration failed\n");
            break;
        }
        waitForReaderResponseAndSendQueued(reader);
        gettimeofday(&Stop, NULL);
        timersub(&Stop, &Start, &Diff);
        Stream->OpenTimeSecs = (double)Diff.tv_usec / 1e6 + Diff.tv_sec;
        MPI_Barrier(Stream->mpiComm);
        gettimeofday(&Stream->ValidStartTime, NULL);
        Stream->RendezvousReaderCount--;
    }
    Stream->Filename = Filename;
    CP_verbose(Stream, "Finish opening Stream \"%s\"\n", Filename);
    AddToLastCallFreeList(Stream);
    return Stream;
}

void sendOneToEachReaderRank(SstStream s, CMFormat f, void *Msg,
                             void **RS_StreamPtr)
{
    for (int i = 0; i < s->ReaderCount; i++)
    {
        WS_ReaderInfo CP_WSR_Stream = s->Readers[i];
        if (CP_WSR_Stream->ReaderStatus == Established)
        {
            CP_verbose(s, "Working on reader cohort %d\n", i);
        }
        else
        {
            CP_verbose(s, "Skipping reader cohort %d\n", i);
            continue;
        }
        sendOneToWSRCohort(CP_WSR_Stream, f, Msg, RS_StreamPtr);
    }
}

static void CP_PeerFailCloseWSReader(WS_ReaderInfo CP_WSR_Stream,
                                     enum StreamStatus NewState);

static void CloseWSRStream(CManager cm, void *WSR_Stream_v)
{
    WS_ReaderInfo CP_WSR_Stream = (WS_ReaderInfo)WSR_Stream_v;
    CP_PeerFailCloseWSReader(CP_WSR_Stream, PeerClosed);
}

static void CP_PeerFailCloseWSReader(WS_ReaderInfo CP_WSR_Stream,
                                     enum StreamStatus NewState)
{
    SstStream ParentStream = CP_WSR_Stream->ParentStream;
    PTHREAD_MUTEX_LOCK(&ParentStream->DataLock);
    if ((NewState == PeerClosed) || (NewState == Closed))
    {
        CP_verbose(ParentStream,
                   "In PeerFailCloseWSReader, releasing timesteps from %ld to "
                   "%ld\n",
                   CP_WSR_Stream->OldestUnreleasedTimestep,
                   CP_WSR_Stream->LastSentTimestep);
        SubRefRangeTimestep(CP_WSR_Stream->ParentStream,
                            CP_WSR_Stream->OldestUnreleasedTimestep,
                            CP_WSR_Stream->LastSentTimestep);
        CP_WSR_Stream->OldestUnreleasedTimestep =
            CP_WSR_Stream->LastSentTimestep + 1;
    }
    if (NewState == PeerFailed)
    {
        CMadd_delayed_task(ParentStream->CPInfo->cm, 2, 0, CloseWSRStream,
                           CP_WSR_Stream);
    }
    CP_verbose(ParentStream, "Moving stream %p to status %s\n", CP_WSR_Stream,
               SSTStreamStatusStr[NewState]);
    CP_WSR_Stream->ReaderStatus = NewState;
    /* main thread might be waiting on timesteps going away */
    pthread_cond_signal(&ParentStream->DataCondition);
    PTHREAD_MUTEX_UNLOCK(&ParentStream->DataLock);
}

void SstWriterClose(SstStream Stream)
{
    struct _WriterCloseMsg Msg;
    struct timeval CloseTime, Diff;
    Msg.FinalTimestep = Stream->LastProvidedTimestep;
    sendOneToEachReaderRank(Stream, Stream->CPInfo->WriterCloseFormat, &Msg,
                            &Msg.RS_Stream);

    KillZeroRefTimesteps(Stream);

    /* wait until all queued data is sent */
    PTHREAD_MUTEX_LOCK(&Stream->DataLock);
    while (Stream->QueuedTimesteps)
    {
        CP_verbose(Stream,
                   "Waiting for timesteps to be released in WriterClose\n");
        if (Stream->Verbose)
        {
            CPTimestepList List = Stream->QueuedTimesteps;
            char *StringList = malloc(1);
            StringList[0] = 0;
            while (List)
            {
                char tmp[20];
                sprintf(tmp, "%ld ", List->Timestep);
                StringList =
                    realloc(StringList, strlen(StringList) + strlen(tmp) + 1);
                strcat(StringList, tmp);
                List = List->Next;
            }
            CP_verbose(Stream, "The timesteps still queued are: %s\n",
                       StringList);
            free(StringList);
        }
        CP_verbose(Stream, "Reader Count is %d\n", Stream->ReaderCount);
        for (int i = 0; i < Stream->ReaderCount; i++)
        {
            CP_verbose(Stream, "Reader [%d] status is %s\n", i,
                       SSTStreamStatusStr[Stream->Readers[i]->ReaderStatus]);
        }
        /* NEED TO HANDLE FAILURE HERE */
        pthread_cond_wait(&Stream->DataCondition, &Stream->DataLock);
    }
    PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);

    gettimeofday(&CloseTime, NULL);
    timersub(&CloseTime, &Stream->ValidStartTime, &Diff);
    if (Stream->Stats)
        Stream->Stats->ValidTimeSecs = (double)Diff.tv_usec / 1e6 + Diff.tv_sec;

    CP_verbose(Stream, "All timesteps are released in WriterClose\n");

    /*
     *  Only rank 0 removes contact info, and only when everything is closed.
     */
    if (Stream->Rank == 0)
    {
        removeContactInfo(Stream);
    }
}

#ifdef NOTDEF
static FFSFormatList AddUniqueFormats(FFSFormatList List,
                                      FFSFormatList Candidates)
{
    while (Candidates)
    {
        FFSFormatList Tmp = List;
        int found = 0;
        while (Tmp)
        {
            if ((Tmp->FormatIDRepLen == Candidates->FormatIDRepLen) &&
                (memcmp(Tmp->FormatIDRep, Candidates->FormatIDRep,
                        Tmp->FormatIDRepLen) == 0))
            {
                found++;
                break;
            }
            Tmp = Tmp->Next;
        }
        if (!found)
        {
            FFSFormatList New = malloc(sizeof(*New));
            memset(New, 0, sizeof(*New));
            New->FormatServerRep = malloc(Candidates->FormatServerRepLen);
            memcpy(New->FormatServerRep, Candidates->FormatServerRep,
                   Candidates->FormatServerRepLen);
            New->FormatServerRepLen = Candidates->FormatServerRepLen;
            New->FormatIDRep = malloc(Candidates->FormatIDRepLen);
            memcpy(New->FormatIDRep, Candidates->FormatIDRep,
                   Candidates->FormatIDRepLen);
            New->FormatIDRepLen = Candidates->FormatIDRepLen;
            New->Next = List;
            List = New;
        }
        Candidates = Candidates->Next;
    }
    return List;
}
#endif

static FFSFormatList AddUniqueFormats(FFSFormatList List,
                                      FFSFormatList Candidates, int copy)
{
    FFSFormatList Tmp = List;
    FFSFormatList Ret = List;

    // If nothing to add, return original
    if (!Candidates)
        return Ret;

    // Add tail of candidates list first
    Ret = AddUniqueFormats(List, Candidates->Next, copy);

    while (Tmp)
    {
        if ((Tmp->FormatIDRepLen == Candidates->FormatIDRepLen) &&
            (memcmp(Tmp->FormatIDRep, Candidates->FormatIDRep,
                    Tmp->FormatIDRepLen) == 0))
        {
            // Identical format already in List, don't add this one
            return Ret;
        }
        Tmp = Tmp->Next;
    }
    // New format not in list, add him to head and return.
    if (copy)
    {
        // Copy top Candidates entry before return
        FFSFormatList Tmp = malloc(sizeof(*Tmp));
        memset(Tmp, 0, sizeof(*Tmp));
        Tmp->FormatServerRep = malloc(Candidates->FormatServerRepLen);
        memcpy(Tmp->FormatServerRep, Candidates->FormatServerRep,
               Candidates->FormatServerRepLen);
        Tmp->FormatServerRepLen = Candidates->FormatServerRepLen;
        Tmp->FormatIDRep = malloc(Candidates->FormatIDRepLen);
        memcpy(Tmp->FormatIDRep, Candidates->FormatIDRep,
               Candidates->FormatIDRepLen);
        Tmp->FormatIDRepLen = Candidates->FormatIDRepLen;
        Candidates = Tmp;
    }
    Candidates->Next = Ret;
    return Candidates;
}

static void DoStreamDiscard(SstStream Stream)
{
    /* No readers, discard earliest timestep in queue (at the end of TS
     * list) */
    if (!Stream->QueuedTimesteps)
    {
        fprintf(stderr, "Expected at least one timestep to dequeue!\n");
    }
    if (Stream->QueuedTimesteps->Next == NULL)
    {
        CPTimestepList Entry = Stream->QueuedTimesteps;
        if (Stream->QueuedTimesteps->ReferenceCount != 0)
        {
            fprintf(stderr,
                    "\n\n\nREFERENCE COUNT CONDITION VIOLATION\n\n\n\n");
        }
        CP_verbose(Stream,
                   "Discarding timestep %d because of queue condition\n",
                   Entry->Timestep);
        Stream->QueuedTimesteps = NULL;
        Entry->FreeTimestep(Entry->FreeClientData);
        free(Entry->Msg);
        free(Entry->DataBlockToFree);
        free(Entry);
        Stream->QueuedTimestepCount--;
    }
    else
    {
        CPTimestepList Last = Stream->QueuedTimesteps;
        while (Last->Next->Next != NULL)
        {
            Last = Last->Next;
        }
        CPTimestepList Entry = Last->Next;
        if (Entry->ReferenceCount != 0)
        {
            fprintf(stderr,
                    "\n\n\nREFERENCE COUNT CONDITION VIOLATION\n\n\n\n");
        }
        CP_verbose(Stream,
                   "Discarding timestep %d because of queue condition\n",
                   Entry->Timestep);
        Last->Next = NULL;
        Entry->FreeTimestep(Entry->FreeClientData);
        free(Entry->MetadataArray);
        free(Entry->DP_TimestepInfo);
        free(Entry->DataBlockToFree);
        free(Entry->Msg);
        free(Entry);
        Stream->QueuedTimestepCount--;
    }
}

static void DoWriterSideGlobalOp(SstStream Stream, int *DiscardIncomingTimestep)
{
    int SendBlockSize = Stream->ReaderCount + 2;
    int *SendBlock = malloc(sizeof(SendBlock[0]) * SendBlockSize);
    int *RecvBlock =
        malloc(sizeof(SendBlock[0]) * SendBlockSize * Stream->CohortSize);

    PTHREAD_MUTEX_LOCK(&Stream->DataLock);
    CP_verbose(Stream, "Initiating Writer Side global operation\n");
    Stream->GlobalOpRequired = 0; //  reset if set
    RequestQueue ArrivingReader = Stream->ReadRequestQueue;

    /* build local info item -- contents:
     *  - Reader_requests_join (integer, contains count, only rank 0 should ever
     *set)
     *  - Current event queue length for this rank
     *	- Reader_status (integer, repeated ReaderCount times)
     */
    SendBlock[0] = 0;
    while (ArrivingReader)
    {
        SendBlock[0]++;
        ArrivingReader = ArrivingReader->Next;
    }
    if (Stream->QueueLimit &&
        (Stream->QueuedTimestepCount > Stream->QueueLimit))
    {
        SendBlock[1] = 1; /* this rank is over stream limit */
    }
    else
    {
        SendBlock[1] = 0; /* No problems here */
    }
    for (int i = 0; i < Stream->ReaderCount; i++)
    {
        SendBlock[i + 2] = Stream->Readers[i]->ReaderStatus;
    }
    MPI_Allgather(SendBlock, SendBlockSize, MPI_INT, RecvBlock, SendBlockSize,
                  MPI_INT, Stream->mpiComm);

    /*
     *  Handle incoming data in a deterministic way.  First handle any changes
     * in state for existing clients.
     */
    int ActiveReaderCount = 0;
    for (int i = 0; i < Stream->ReaderCount; i++)
    {

        if (Stream->Readers[i]->ReaderStatus == Established)
        {
            /*
             * if we think a reader is OK, see if anyone has had a failure with
             * it
             */
            for (int j = 0; j < Stream->CohortSize; j++)
            {
                if (RecvBlock[j * SendBlockSize + i + 2] == PeerFailed)
                {
                    Stream->Readers[i]->ReaderStatus = PeerFailed;
                }
            }
        }
        if (Stream->Readers[i]->ReaderStatus == PeerFailed)
        {
            // If we see PeerFailed now, everyone does, move to Fail
            CP_verbose(Stream, "Reader %d now determined to have failed, "
                               "dereferencing the timesteps it was sent, %d to "
                               "%d\n",
                       i, Stream->Readers[i]->StartingTimestep,
                       Stream->Readers[i]->LastSentTimestep);
            PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
            CP_PeerFailCloseWSReader(Stream->Readers[i], Closed);
            PTHREAD_MUTEX_LOCK(&Stream->DataLock);
        }
        if (Stream->Readers[i]->ReaderStatus == Established)
            ActiveReaderCount++;
    }

    /*
     *  Then handle possible incoming connection requests.  (Only rank 0 has
     * valid info.)
     */
    int OverLimit = 0;
    for (int i = 0; i < Stream->CohortSize; i++)
    {
        if (RecvBlock[i * SendBlockSize + 1] > 0)
            OverLimit++;
    }
    /* we've got all the state we need to know, release data lock */
    PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);

    for (int i = 0; i < RecvBlock[0]; i++)
    {
        WS_ReaderInfo reader;
        CP_verbose(Stream,
                   "Writer side Global operation accepting incoming reader\n");
        reader = WriterParticipateInReaderOpen(Stream);
        if (!reader)
        {
            CP_error(Stream, "Potential reader registration failed\n");
            break;
        }
        waitForReaderResponseAndSendQueued(reader);
    }

    /*
     *  Lastly, we'll decide what to do with the current provided timestep.
     *  If any writer rank is over the queuelimit, we must discard or block
     * decision points:
     If mode is block on queue limit:
         If there are no currently active readers, every node must block on
     reader arrival.
         else if there are active readers, each rank can block on queue size
     decreasing (as it will with progress)
     if mode is discard:
          If there are no currently active readers, we can discard the oldest
     item in the queue
          else if there are active readers, we must discard the currently
     arriving timestep.
     */
    if (OverLimit)
    {
        if (Stream->QueueFullPolicy == SstQueueFullBlock)
        {
            if (ActiveReaderCount == 0)
            {
                while (ActiveReaderCount == 0)
                {
                    WS_ReaderInfo reader;
                    if (Stream->Rank == 0)
                    {
                        PTHREAD_MUTEX_LOCK(&Stream->DataLock);
                        while (Stream->ReadRequestQueue == NULL)
                        {
                            pthread_cond_wait(&Stream->DataCondition,
                                              &Stream->DataLock);
                        }
                        PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
                    }
                    MPI_Barrier(Stream->mpiComm);
                    CP_verbose(Stream, "Writer side Global operation blocking "
                                       "for incoming reader\n");
                    reader = WriterParticipateInReaderOpen(Stream);
                    if (!reader)
                    {
                        CP_error(Stream,
                                 "Potential reader registration failed\n");
                        break;
                    }
                    waitForReaderResponseAndSendQueued(reader);
                    ActiveReaderCount++;
                }
            }
            else
            {
                /* wait until enough queued data is sent */
                PTHREAD_MUTEX_LOCK(&Stream->DataLock);
                while (Stream->QueueLimit &&
                       (Stream->QueuedTimestepCount > Stream->QueueLimit))
                {
                    CP_verbose(
                        Stream,
                        "Waiting for timesteps to be released in GlobalOp\n");
                    CP_verbose(Stream,
                               "The first timestep still queued is %d\n",
                               Stream->QueuedTimesteps->Timestep);
                    /* NEED TO HANDLE FAILURE HERE */
                    pthread_cond_wait(&Stream->DataCondition,
                                      &Stream->DataLock);
                }
                PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
            }
        }
        else
        {
            /* discard things */
            if (ActiveReaderCount == 0)
            {
                DoStreamDiscard(Stream);
            }
            else
            {
                *DiscardIncomingTimestep = 1;
            }
        }
    }
    free(SendBlock);
    free(RecvBlock);
    CP_verbose(Stream, "Finished with Writer side Global operations\n");
}

extern void SstInternalProvideTimestep(
    SstStream Stream, SstData LocalMetadata, SstData Data, long Timestep,
    FFSFormatList Formats, DataFreeFunc FreeTimestep, void *FreeClientData,
    SstData AttributeData, DataFreeFunc FreeAttributeData,
    void *FreeAttributelientData)
{
    void *data_block;
    MetadataPlusDPInfo *pointers;
    struct _TimestepMetadataMsg *Msg = malloc(sizeof(*Msg));
    void *DP_TimestepInfo = NULL;
    struct _MetadataPlusDPInfo Md;
    CPTimestepList Entry = calloc(1, sizeof(struct _CPTimestepEntry));
    FFSFormatList XmitFormats = NULL;
    int GlobalOpRequested = 0;
    int GlobalOpRequired = 0;

    memset(Msg, 0, sizeof(*Msg));
    PTHREAD_MUTEX_LOCK(&Stream->DataLock);<commit_msg>Clean up queue handling in GlobalOp to produce closer to expected results for Discard on queue full<commit_after>#include <assert.h>
#include <limits.h>
#include <signal.h>
#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <unistd.h>

#include "adios2/ADIOSConfig.h"
#include <atl.h>
#include <evpath.h>
#ifdef ADIOS2_HAVE_MPI
#include <mpi.h>
#else
#include "sstmpidummy.h"
#endif
#include <pthread.h>

#include "sst.h"

#include "cp_internal.h"

extern void CP_verbose(SstStream Stream, char *Format, ...);

static void sendOneToEachWriterRank(SstStream s, CMFormat f, void *Msg,
                                    void **WS_StreamPtr);
static void CP_PeerFailCloseWSReader(WS_ReaderInfo CP_WSR_Stream,
                                     enum StreamStatus NewState);

#ifdef MUTEX_DEBUG
#define PTHREAD_MUTEX_LOCK(lock)                                               \
    printf("Trying lock line %d\n", __LINE__);                                 \
    pthread_mutex_lock(lock);                                                  \
    printf("Got lock\n");
#define PTHREAD_MUTEX_UNLOCK(lock)                                             \
    printf("UNlocking line %d\n", __LINE__);                                   \
    pthread_mutex_unlock(lock);
#else
#define PTHREAD_MUTEX_LOCK(lock) pthread_mutex_lock(lock);
#define PTHREAD_MUTEX_UNLOCK(lock) pthread_mutex_unlock(lock);
#endif

static char *buildContactInfo(SstStream Stream)
{
    char *Contact = CP_GetContactString(Stream);
    char *FullInfo = malloc(strlen(Contact) + 20);
    sprintf(FullInfo, "%p:%s", (void *)Stream, Contact);
    free(Contact);
    return FullInfo;
}

struct NameListEntry
{
    const char *FileName;
    struct NameListEntry *Next;
};

struct NameListEntry *FileNameList = NULL;

static void RemoveAllFilesInList()
{
    while (FileNameList)
    {
        struct NameListEntry *Next = FileNameList->Next;
        fprintf(stderr, "SST stream open at exit, unlinking contact file %s\n",
                FileNameList->FileName);
        unlink(FileNameList->FileName);
        free(FileNameList);
        FileNameList = Next;
    }
}

static void ExitAndRemoveFiles(int signum)
{
    fprintf(stderr, "ADIOS2 caught SigInt, exiting with error\n");
    exit(1);
}

static void AddNameToExitList(const char *FileName)
{
    static int First = 1;
    if (First)
    {
        struct sigaction sa;
        First = 0;
        atexit(RemoveAllFilesInList);

        memset(&sa, 0, sizeof(sa));
        sa.sa_handler = ExitAndRemoveFiles;
        sigemptyset(&sa.sa_mask);
        sigaction(SIGINT, &sa, NULL);
    }

    struct NameListEntry *NewHead = malloc(sizeof(*NewHead));
    NewHead->FileName = FileName;
    NewHead->Next = FileNameList;
    FileNameList = NewHead;
}

static void RemoveNameFromExitList(const char *FileName)
{
    struct NameListEntry **LastPtr = &FileNameList;
    struct NameListEntry *List = FileNameList;
    while (List)
    {
        if (strcmp(FileName, List->FileName) == 0)
        {
            *LastPtr = List->Next;
            free(List);
            return;
        }
        List = List->Next;
    }
}

static void writeContactInfoFile(const char *Name, SstStream Stream)
{
    char *Contact = buildContactInfo(Stream);
    char *TmpName = malloc(strlen(Name) + strlen(".tmp") + 1);
    char *FileName = malloc(strlen(Name) + strlen(SST_POSTFIX) + 1);
    FILE *WriterInfo;

    /*
     * write the contact information file with a temporary name before
     * renaming it to the final version to help prevent partial reads
     */
    sprintf(TmpName, "%s.tmp", Name);
    sprintf(FileName, "%s" SST_POSTFIX, Name);
    WriterInfo = fopen(TmpName, "w");
    fprintf(WriterInfo, "%s", Contact);
    fclose(WriterInfo);
    rename(TmpName, FileName);
    Stream->AbsoluteFilename = realpath(FileName, NULL);
    free(Contact);
    free(TmpName);
    free(FileName);
    AddNameToExitList(Stream->AbsoluteFilename);
}

static void writeContactInfoScreen(const char *Name, SstStream Stream)
{
    char *Contact = buildContactInfo(Stream);

    /*
     * write the contact information file to the screen
     */
    fprintf(stdout, "The next line of output is the contact information "
                    "associated with SST output stream \"%s\".  Please make it "
                    "available to the reader.\n",
            Name);
    fprintf(stdout, "\t%s\n", Contact);
    free(Contact);
}

static void registerContactInfo(const char *Name, SstStream Stream)
{
    switch (Stream->RegistrationMethod)
    {
    case SstRegisterFile:
        writeContactInfoFile(Name, Stream);
        break;
    case SstRegisterScreen:
        writeContactInfoScreen(Name, Stream);
        break;
    case SstRegisterCloud:
        /* not yet */
        break;
    }
}

static void removeContactInfoFile(SstStream Stream)
{
    const char *Name = Stream->AbsoluteFilename;
    unlink(Name);
    RemoveNameFromExitList(Name);
}

static void removeContactInfo(SstStream Stream)
{
    switch (Stream->RegistrationMethod)
    {
    case SstRegisterFile:
        removeContactInfoFile(Stream);
        break;
    case SstRegisterScreen:
        /* nothing necessary here */
        break;
    case SstRegisterCloud:
        /* not yet */
        break;
    }
}

static void WriterConnCloseHandler(CManager cm, CMConnection closed_conn,
                                   void *client_data)
{
    WS_ReaderInfo WSreader = (WS_ReaderInfo)client_data;
    SstStream ParentWriterStream = WSreader->ParentStream;

    PTHREAD_MUTEX_LOCK(&ParentWriterStream->DataLock);
    if (WSreader->ReaderStatus == Established)
    {
        /*
         * tag our reader instance as failed.
         * If any instance is failed, we should remove all, but that requires a
         * global operation, so prep.
         */
        CP_verbose(ParentWriterStream, "Writer-side Rank received a "
                                       "connection-close event during normal "
                                       "operations, peer likely failed\n");
        PTHREAD_MUTEX_UNLOCK(&ParentWriterStream->DataLock);
        CP_PeerFailCloseWSReader(WSreader, PeerFailed);
        PTHREAD_MUTEX_LOCK(&ParentWriterStream->DataLock);
        ParentWriterStream->GlobalOpRequired = 1;
    }
    else if ((WSreader->ReaderStatus == PeerClosed) ||
             (WSreader->ReaderStatus == Closed))
    {
        /* ignore this.  We expect a close after the connection is marked closed
         */
        CP_verbose(ParentWriterStream, "Writer-side Rank received a "
                                       "connection-close event after close, "
                                       "not unexpected\n");
    }
    else
    {
        fprintf(stderr, "Got an unexpected connection close event\n");
        CP_verbose(ParentWriterStream, "Writer-side Rank received a "
                                       "connection-close event in unexpected "
                                       "state %s\n",
                   SSTStreamStatusStr[WSreader->ReaderStatus]);
        PTHREAD_MUTEX_UNLOCK(&ParentWriterStream->DataLock);
        CP_PeerFailCloseWSReader(WSreader, PeerFailed);
        PTHREAD_MUTEX_LOCK(&ParentWriterStream->DataLock);
    }
    /* main thread might be waiting on status change */
    pthread_cond_signal(&ParentWriterStream->DataCondition);
    PTHREAD_MUTEX_UNLOCK(&ParentWriterStream->DataLock);
}

static void SendPeerSetupMsg(WS_ReaderInfo reader, int reversePeer, int myRank)
{
    CMConnection conn = reader->Connections[reversePeer].CMconn;
    SstStream Stream = reader->ParentStream;
    struct _PeerSetupMsg setup;
    setup.RS_Stream = reader->Connections[reversePeer].RemoteStreamID;
    setup.WriterRank = myRank;
    setup.WriterCohortSize = Stream->CohortSize;
    if (CMwrite(conn, Stream->CPInfo->PeerSetupFormat, &setup) != 1)
    {
        CP_verbose(Stream,
                   "Message failed to send to reader in sendPeerSetup in "
                   "reader open\n");
    }
}

static int initWSReader(WS_ReaderInfo reader, int ReaderSize,
                        CP_ReaderInitInfo *reader_info)
{
    int WriterSize = reader->ParentStream->CohortSize;
    int WriterRank = reader->ParentStream->Rank;
    int i;
    int *reverseArray;
    reader->ReaderCohortSize = ReaderSize;
    if (!reader->Connections)
    {
        reader->Connections =
            calloc(sizeof(reader->Connections[0]), ReaderSize);
    }
    for (i = 0; i < ReaderSize; i++)
    {
        if (!reader->Connections[i].ContactList)
        {
            reader->Connections[i].ContactList =
                attr_list_from_string(reader_info[i]->ContactInfo);
        }
        reader->Connections[i].RemoteStreamID = reader_info[i]->ReaderID;
        reader->Connections[i].CMconn = NULL;
    }
    /*
     *   Peering.
     *   We use peering for two things:
     *     - failure awareness (each rank needs a close handler on one
     connection to some opposite rank so they can detect failure)
     *     - notification (how info gets sent from reader to writer and vice
     versa)
     *
     *   A connection that exists for notification is also useful for
     *   failure awareness, but where not are necessary for
     *   notification, we still may make some for failure
     *   notification.

     *   In this code, all connections are made by the writing side,
     *   but the reader side must be sent notifications so that it is
     *   aware of what connections are made for it and what they are
     *   to be used for (I.E. notification, or only existing passively
     *   for failure notification).
     *
     *   Connections that are used for notification from writer to
     *   reader will be in the Peer list and we'll send messages down
     *   them later.  If there are many more writers than readers
     *   (presumed normal case), the peer list will have 0 or 1
     *   entries.  Connections in the reverseArray are for failure
     *   awareness and/or notification from reader to writer.  If
     *   there are many more readers than writers, the reverseArray
     *   will have one entry (to the one reader that will send us
     *   notifications and which we will use for failure awareness).

     *   If there are equal numbers of readers and writers, then each
     *   rank is peered only with the same rank in the opposing.

     *   If there happen to be many more readers than writers, then
     *   the Peer list will contain a lot of entries (all those that
     *   get notifications from us.  The reverseArray will also
     *   contain a lot of entries, but only the first will send us
     *   notifications.  The others will just use the connections for
     *   failure awareness.

     */
    getPeerArrays(WriterSize, WriterRank, ReaderSize, &reader->Peers,
                  &reverseArray);

    i = 0;
    while (reverseArray[i] != -1)
    {
        int peer = reverseArray[i];
        if (reader->ParentStream->ConnectionUsleepMultiplier != 0)
            usleep(WriterRank *
                   reader->ParentStream->ConnectionUsleepMultiplier);
        reader->Connections[peer].CMconn =
            CMget_conn(reader->ParentStream->CPInfo->cm,
                       reader->Connections[peer].ContactList);

        if (!reader->Connections[peer].CMconn)
        {
            CP_error(
                reader->ParentStream,
                "Connection failed in SstInitWSReader! Contact list was:\n");
            CP_error(
                reader->ParentStream, "%s\n",
                attr_list_to_string(reader->Connections[peer].ContactList));
            /* fail the stream */
            return 0;
        }

        CMconn_register_close_handler(reader->Connections[peer].CMconn,
                                      WriterConnCloseHandler, (void *)reader);
        if (i == 0)
        {
            /* failure awareness for reader rank */
            CP_verbose(reader->ParentStream, "Sending peer setup to rank %d\n",
                       peer);
            SendPeerSetupMsg(reader, peer, reader->ParentStream->Rank);
        }
        else
        {
            CP_verbose(reader->ParentStream, "Sending peer setup to rank %d\n",
                       peer);
            /* failure awareness for reader rank */
            SendPeerSetupMsg(reader, peer, -1);
        }
        i++;
    }
    free(reverseArray);
    i = 0;
    while (reader->Peers[i] != -1)
    {
        int peer = reader->Peers[i];
        if (reader->Connections[peer].CMconn)
        {
            /* already made this above */
            i++;
            continue;
        }
        if (reader->ParentStream->ConnectionUsleepMultiplier != 0)
            usleep(WriterRank *
                   reader->ParentStream->ConnectionUsleepMultiplier);
        reader->Connections[peer].CMconn =
            CMget_conn(reader->ParentStream->CPInfo->cm,
                       reader->Connections[peer].ContactList);

        if (!reader->Connections[peer].CMconn)
        {
            CP_error(
                reader->ParentStream,
                "Connection failed in SstInitWSReader! Contact list was:\n");
            CP_error(
                reader->ParentStream, "%s\n",
                attr_list_to_string(reader->Connections[peer].ContactList));
            /* fail the stream */
            return 0;
        }

        CMconn_register_close_handler(reader->Connections[peer].CMconn,
                                      WriterConnCloseHandler, (void *)reader);
        /* failure awareness for reader rank */
        CP_verbose(reader->ParentStream, "Sending peer setup to rank %d\n",
                   peer);
        SendPeerSetupMsg(reader, peer, reader->ParentStream->Rank);
        i++;
    }
    return 1;
}

static long earliestAvailableTimestepNumber(SstStream Stream,
                                            long CurrentTimestep)
{
    long Ret = CurrentTimestep;
    CPTimestepList List = Stream->QueuedTimesteps;
    PTHREAD_MUTEX_LOCK(&Stream->DataLock);
    while (List)
    {
        if (List->Timestep < Ret)
        {
            List->ReferenceCount++;
            Ret = List->Timestep;
            CP_verbose(Stream, "Earliest available : Writer-side Timestep %ld "
                               "now has reference count %d\n",
                       List->Timestep, List->ReferenceCount);
        }
        List = List->Next;
    }
    PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
    return Ret;
}

static void SubRefRangeTimestep(SstStream Stream, long LowRange, long HighRange)
{
    CPTimestepList Last = NULL, List;
    int AnythingRemoved = 0;
    List = Stream->QueuedTimesteps;
    while (List)
    {
        if ((List->Timestep >= LowRange) && (List->Timestep <= HighRange))
        {
            List->ReferenceCount--;
            CP_verbose(Stream, "SubRef : Writer-side Timestep %ld now has "
                               "reference count %d\n",
                       List->Timestep, List->ReferenceCount);
        }
        if (List->ReferenceCount == 0)
        {
            /* dequeue and free */
            CPTimestepList ItemToFree = List;
            if (Last == NULL)
            {
                Stream->QueuedTimesteps = List->Next;
            }
            else
            {
                Last->Next = List->Next;
            }
            CP_verbose(Stream, "Step %d reference count reached zero, "
                               "releasing from DP and freeing\n",
                       ItemToFree->Timestep);
            Stream->DP_Interface->releaseTimestep(&Svcs, Stream->DP_Stream,
                                                  List->Timestep);

            Stream->QueuedTimestepCount--;
            List = List->Next;
            ItemToFree->FreeTimestep(ItemToFree->FreeClientData);
            free(ItemToFree->Msg);
            free(ItemToFree->MetadataArray);
            free(ItemToFree->DP_TimestepInfo);
            free(ItemToFree->DataBlockToFree);
            free(ItemToFree);
            AnythingRemoved++;
        }
        else
        {
            Last = List;
            List = List->Next;
        }
    }
    if (AnythingRemoved)
    {
        /* main thread might be waiting on timesteps going away */
        pthread_cond_signal(&Stream->DataCondition);
    }
}

static void KillZeroRefTimesteps(SstStream Stream)
{
    CPTimestepList Last = NULL, List;
    int AnythingRemoved = 0;
    PTHREAD_MUTEX_LOCK(&Stream->DataLock);
    List = Stream->QueuedTimesteps;
    while (List)
    {
        if (List->ReferenceCount == 0)
        {
            /* dequeue and free */
            CPTimestepList ItemToFree = List;
            if (Last == NULL)
            {
                Stream->QueuedTimesteps = List->Next;
            }
            else
            {
                Last->Next = List->Next;
            }
            CP_verbose(Stream, "Step %d reference count is zero, "
                               "releasing from DP and freeing\n",
                       List->Timestep);
            Stream->DP_Interface->releaseTimestep(&Svcs, Stream->DP_Stream,
                                                  List->Timestep);

            Stream->QueuedTimestepCount--;
            List = List->Next;
            free(ItemToFree);
            AnythingRemoved++;
        }
        else
        {
            Last = List;
            List = List->Next;
        }
    }
    if (AnythingRemoved)
    {
        /* main thread might be waiting on timesteps going away */
        pthread_cond_signal(&Stream->DataCondition);
    }
    PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
}

WS_ReaderInfo WriterParticipateInReaderOpen(SstStream Stream)
{
    RequestQueue Req;
    reader_data_t ReturnData;
    void *free_block = NULL;
    int WriterResponseCondition = -1;
    CMConnection conn;
    long MyStartingTimestep, GlobalStartingTimestep;

    CP_verbose(Stream, "Beginning writer-side reader open protocol\n");
    if (Stream->Rank == 0)
    {
        PTHREAD_MUTEX_LOCK(&Stream->DataLock);
        assert((Stream->ReadRequestQueue));
        Req = Stream->ReadRequestQueue;
        Stream->ReadRequestQueue = Req->Next;
        Req->Next = NULL;
        PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
        struct _CombinedReaderInfo reader_data;
        memset(&reader_data, 0, sizeof(reader_data));
        reader_data.ReaderCohortSize = Req->Msg->ReaderCohortSize;
        reader_data.CP_ReaderInfo = Req->Msg->CP_ReaderInfo;
        reader_data.DP_ReaderInfo = Req->Msg->DP_ReaderInfo;
        ReturnData = CP_distributeDataFromRankZero(
            Stream, &reader_data, Stream->CPInfo->CombinedReaderInfoFormat,
            &free_block);
        WriterResponseCondition = Req->Msg->WriterResponseCondition;
        conn = Req->Conn;
        CMreturn_buffer(Stream->CPInfo->cm, Req->Msg);
        free(Req);
    }
    else
    {
        ReturnData = CP_distributeDataFromRankZero(
            Stream, NULL, Stream->CPInfo->CombinedReaderInfoFormat,
            &free_block);
    }
    //    printf("I am writer rank %d, my info on readers is:\n", Stream->Rank);
    //    FMdump_data(FMFormat_of_original(Stream->CPInfo->combined_reader_Format),
    //                ReturnData, 1024000);
    //    printf("\n");

    Stream->Readers = realloc(Stream->Readers, sizeof(Stream->Readers[0]) *
                                                   (Stream->ReaderCount + 1));
    DP_WSR_Stream per_reader_Stream;
    void *DP_WriterInfo;
    void *ret_data_block;
    CP_PeerConnection *connections_to_reader;
    connections_to_reader =
        calloc(sizeof(CP_PeerConnection), ReturnData->ReaderCohortSize);
    for (int i = 0; i < ReturnData->ReaderCohortSize; i++)
    {
        attr_list attrs =
            attr_list_from_string(ReturnData->CP_ReaderInfo[i]->ContactInfo);
        connections_to_reader[i].ContactList = attrs;
        connections_to_reader[i].RemoteStreamID =
            ReturnData->CP_ReaderInfo[i]->ReaderID;
    }

    per_reader_Stream = Stream->DP_Interface->initWriterPerReader(
        &Svcs, Stream->DP_Stream, ReturnData->ReaderCohortSize,
        connections_to_reader, ReturnData->DP_ReaderInfo, &DP_WriterInfo);

    WS_ReaderInfo CP_WSR_Stream = malloc(sizeof(*CP_WSR_Stream));
    CP_WSR_Stream->ReaderStatus = NotOpen;
    Stream->Readers[Stream->ReaderCount] = CP_WSR_Stream;
    CP_WSR_Stream->DP_WSR_Stream = per_reader_Stream;
    CP_WSR_Stream->ParentStream = Stream;
    CP_WSR_Stream->Connections = connections_to_reader;

    int MySuccess = initWSReader(CP_WSR_Stream, ReturnData->ReaderCohortSize,
                                 ReturnData->CP_ReaderInfo);

    int GlobalSuccess = 0;
    MPI_Allreduce(&MySuccess, &GlobalSuccess, 1, MPI_INT, MPI_LAND,
                  Stream->mpiComm);

    if (!GlobalSuccess)
    {
        return NULL;
    }
    AddToLastCallFreeList(CP_WSR_Stream);
    free(free_block);
    ReturnData = NULL; /* now invalid */

    Stream->ReaderCount++;

    struct _CP_DP_PairInfo combined_init;
    struct _CP_WriterInitInfo cpInfo;

    struct _CP_DP_PairInfo **pointers = NULL;

    memset(&cpInfo, 0, sizeof(cpInfo));
    cpInfo.ContactInfo = CP_GetContactString(Stream);
    cpInfo.WriterID = CP_WSR_Stream;

    combined_init.CP_Info = (void **)&cpInfo;
    combined_init.DP_Info = DP_WriterInfo;

    MyStartingTimestep =
        earliestAvailableTimestepNumber(Stream, Stream->WriterTimestep);
    if (MyStartingTimestep == -1)
        MyStartingTimestep = 0;

    MPI_Allreduce(&MyStartingTimestep, &GlobalStartingTimestep, 1, MPI_LONG,
                  MPI_MAX, Stream->mpiComm);

    /*
     *  The earliestAvailableTimestepNumber subroutine also added to the
     *  reference count of all existing timesteps, in case we will be making
     *  it available to the joining reader.  Once we've determined what the
     *  starting timestep we'll make available is (GlobalStartingTimestep),
     *  decrement the ref count of others.
     */
    if (MyStartingTimestep != GlobalStartingTimestep)
    {
        CP_verbose(Stream, "In writer participate in reader open, releasing "
                           "timesteps from %ld to %ld\n",
                   MyStartingTimestep, GlobalStartingTimestep - 1);

        SubRefRangeTimestep(Stream, MyStartingTimestep,
                            GlobalStartingTimestep - 1);
    }
    CP_verbose(Stream,
               "My oldest timestep was %ld, global oldest timestep was %ld\n",
               MyStartingTimestep, GlobalStartingTimestep);

    CP_WSR_Stream->StartingTimestep = GlobalStartingTimestep;

    pointers = (struct _CP_DP_PairInfo **)CP_consolidateDataToRankZero(
        Stream, &combined_init, Stream->CPInfo->PerRankWriterInfoFormat,
        &ret_data_block);

    if (Stream->Rank == 0)
    {
        struct _WriterResponseMsg response;
        memset(&response, 0, sizeof(response));
        response.WriterResponseCondition = WriterResponseCondition;
        response.WriterCohortSize = Stream->CohortSize;
        response.WriterConfigParams = Stream->ConfigParams;
        response.NextStepNumber = GlobalStartingTimestep;
        response.CP_WriterInfo =
            malloc(response.WriterCohortSize * sizeof(void *));
        response.DP_WriterInfo =
            malloc(response.WriterCohortSize * sizeof(void *));
        for (int i = 0; i < response.WriterCohortSize; i++)
        {
            response.CP_WriterInfo[i] =
                (struct _CP_WriterInitInfo *)pointers[i]->CP_Info;
            response.DP_WriterInfo[i] = pointers[i]->DP_Info;
        }
        if (CMwrite(conn, Stream->CPInfo->WriterResponseFormat, &response) != 1)
        {
            CP_verbose(Stream,
                       "Message failed to send to reader in participate in "
                       "reader open\n");
        }
        free(response.CP_WriterInfo);
        free(response.DP_WriterInfo);
    }
    free(cpInfo.ContactInfo);
    if (ret_data_block)
        free(ret_data_block);
    if (pointers)
        free(pointers);
    Stream->NewReaderPresent = 1;
    CP_verbose(Stream, "Finish writer-side reader open protocol for reader %p, "
                       "reader ready response pending\n",
               CP_WSR_Stream);
    return CP_WSR_Stream;
}

void sendOneToWSRCohort(WS_ReaderInfo CP_WSR_Stream, CMFormat f, void *Msg,
                        void **RS_StreamPtr)
{
    SstStream s = CP_WSR_Stream->ParentStream;
    int j = 0;

    while (CP_WSR_Stream->Peers[j] != -1)
    {
        int peer = CP_WSR_Stream->Peers[j];
        CMConnection conn = CP_WSR_Stream->Connections[peer].CMconn;
        /* add the reader-rank-specific Stream identifier to each outgoing
         * message */
        *RS_StreamPtr = CP_WSR_Stream->Connections[peer].RemoteStreamID;
        CP_verbose(s, "Sending a message to reader %d (%p)\n", peer,
                   *RS_StreamPtr);
        if (CMwrite(conn, f, Msg) != 1)
        {
            CP_verbose(s, "Message failed to send to reader %d (%p)\n", peer,
                       *RS_StreamPtr);
        }
        j++;
    }
}

static void waitForReaderResponseAndSendQueued(WS_ReaderInfo Reader)
{
    SstStream Stream = Reader->ParentStream;
    PTHREAD_MUTEX_LOCK(&Stream->DataLock);
    while (Reader->ReaderStatus != Established)
    {
        /* NEED TO HANDLE FAILURE HERE */
        CP_verbose(Stream, "Waiting for Reader ready on WSR %p.\n", Reader);
        pthread_cond_wait(&Stream->DataCondition, &Stream->DataLock);
    }
    assert(Reader->ReaderStatus == Established);
    PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);

    /* send any queued metadata necessary */
    Reader->OldestUnreleasedTimestep = Reader->StartingTimestep;
    CP_verbose(Stream, "Reader ready on WSR %p, Stream established.\n", Reader);
    for (long TS = Reader->StartingTimestep; TS <= Stream->LastProvidedTimestep;
         TS++)
    {
        CPTimestepList List = Stream->QueuedTimesteps;
        while (List)
        {
            if (List->Timestep == TS)
            {
                FFSFormatList SavedFormats = List->Msg->Formats;
                CP_verbose(Stream,
                           "Sending Queued TimestepMetadata for timestep %d\n",
                           TS);

                if (TS == Reader->StartingTimestep)
                {
                    /* For first Msg, send all previous formats */
                    List->Msg->Formats = Stream->PreviousFormats;
                }
                else
                {
                    /*
                     *  TENTATIVE!  TRYING TO SEE IF THIS MIGHT IMPACT RARE
                     *  STUCK READER PROBLEM.
                     *  Add  a short delay between consecutive messages
                     */
                    usleep(10 * Stream->ConnectionUsleepMultiplier);
                }
                sendOneToWSRCohort(
                    Reader, Stream->CPInfo->DeliverTimestepMetadataFormat,
                    List->Msg, &List->Msg->RS_Stream);
                if (TS == Reader->StartingTimestep)
                {
                    /* restore Msg format list */
                    List->Msg->Formats = SavedFormats;
                }
            }
            List = List->Next;
        }
    }
}

SstStream SstWriterOpen(const char *Name, SstParams Params, MPI_Comm comm)
{
    SstStream Stream;

    Stream = CP_newStream();
    Stream->Role = WriterRole;
    CP_validateParams(Stream, Params, 1 /* Writer */);
    Stream->ConfigParams = Params;

    char *Filename = strdup(Name);

    Stream->mpiComm = comm;

    MPI_Comm_rank(Stream->mpiComm, &Stream->Rank);
    MPI_Comm_size(Stream->mpiComm, &Stream->CohortSize);

    Stream->DP_Interface = SelectDP(&Svcs, Stream, Stream->ConfigParams);

    if (!Stream->DP_Interface)
    {
        CP_verbose(Stream, "Failed to load DataPlane %s for Stream \"%s\"\n",
                   Params->DataTransport, Filename);
        return NULL;
    }

    Stream->CPInfo = CP_getCPInfo(Stream->DP_Interface);

    if (Stream->RendezvousReaderCount > 0)
    {
        Stream->FirstReaderCondition =
            CMCondition_get(Stream->CPInfo->cm, NULL);
    }
    else
    {
        Stream->FirstReaderCondition = -1;
    }

    Stream->DP_Stream =
        Stream->DP_Interface->initWriter(&Svcs, Stream, Stream->ConfigParams);

    if (Stream->Rank == 0)
    {
        registerContactInfo(Filename, Stream);
    }

    CP_verbose(Stream, "Opening Stream \"%s\"\n", Filename);

    if (Stream->Rank == 0)
    {
        CP_verbose(Stream, "Writer stream params are:\n");
        CP_dumpParams(Stream, Stream->ConfigParams);
    }

    while (Stream->RendezvousReaderCount > 0)
    {
        WS_ReaderInfo reader;
        CP_verbose(Stream, "Stream \"%s\" waiting for %d readers\n", Filename,
                   Stream->RendezvousReaderCount);
        if (Stream->Rank == 0)
        {
            PTHREAD_MUTEX_LOCK(&Stream->DataLock);
            if (Stream->ReadRequestQueue == NULL)
            {
                pthread_cond_wait(&Stream->DataCondition, &Stream->DataLock);
            }
            assert(Stream->ReadRequestQueue);
            PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
        }
        MPI_Barrier(Stream->mpiComm);

        struct timeval Start, Stop, Diff;
        gettimeofday(&Start, NULL);
        reader = WriterParticipateInReaderOpen(Stream);
        if (!reader)
        {
            CP_error(Stream, "Potential reader registration failed\n");
            break;
        }
        waitForReaderResponseAndSendQueued(reader);
        gettimeofday(&Stop, NULL);
        timersub(&Stop, &Start, &Diff);
        Stream->OpenTimeSecs = (double)Diff.tv_usec / 1e6 + Diff.tv_sec;
        MPI_Barrier(Stream->mpiComm);
        gettimeofday(&Stream->ValidStartTime, NULL);
        Stream->RendezvousReaderCount--;
    }
    Stream->Filename = Filename;
    CP_verbose(Stream, "Finish opening Stream \"%s\"\n", Filename);
    AddToLastCallFreeList(Stream);
    return Stream;
}

void sendOneToEachReaderRank(SstStream s, CMFormat f, void *Msg,
                             void **RS_StreamPtr)
{
    for (int i = 0; i < s->ReaderCount; i++)
    {
        WS_ReaderInfo CP_WSR_Stream = s->Readers[i];
        if (CP_WSR_Stream->ReaderStatus == Established)
        {
            CP_verbose(s, "Working on reader cohort %d\n", i);
        }
        else
        {
            CP_verbose(s, "Skipping reader cohort %d\n", i);
            continue;
        }
        sendOneToWSRCohort(CP_WSR_Stream, f, Msg, RS_StreamPtr);
    }
}

static void CP_PeerFailCloseWSReader(WS_ReaderInfo CP_WSR_Stream,
                                     enum StreamStatus NewState);

static void CloseWSRStream(CManager cm, void *WSR_Stream_v)
{
    WS_ReaderInfo CP_WSR_Stream = (WS_ReaderInfo)WSR_Stream_v;
    CP_PeerFailCloseWSReader(CP_WSR_Stream, PeerClosed);
}

static void CP_PeerFailCloseWSReader(WS_ReaderInfo CP_WSR_Stream,
                                     enum StreamStatus NewState)
{
    SstStream ParentStream = CP_WSR_Stream->ParentStream;
    PTHREAD_MUTEX_LOCK(&ParentStream->DataLock);
    if ((NewState == PeerClosed) || (NewState == Closed))
    {
        CP_verbose(ParentStream,
                   "In PeerFailCloseWSReader, releasing timesteps from %ld to "
                   "%ld\n",
                   CP_WSR_Stream->OldestUnreleasedTimestep,
                   CP_WSR_Stream->LastSentTimestep);
        SubRefRangeTimestep(CP_WSR_Stream->ParentStream,
                            CP_WSR_Stream->OldestUnreleasedTimestep,
                            CP_WSR_Stream->LastSentTimestep);
        CP_WSR_Stream->OldestUnreleasedTimestep =
            CP_WSR_Stream->LastSentTimestep + 1;
    }
    if (NewState == PeerFailed)
    {
        CMadd_delayed_task(ParentStream->CPInfo->cm, 2, 0, CloseWSRStream,
                           CP_WSR_Stream);
    }
    CP_verbose(ParentStream, "Moving stream %p to status %s\n", CP_WSR_Stream,
               SSTStreamStatusStr[NewState]);
    CP_WSR_Stream->ReaderStatus = NewState;
    /* main thread might be waiting on timesteps going away */
    pthread_cond_signal(&ParentStream->DataCondition);
    PTHREAD_MUTEX_UNLOCK(&ParentStream->DataLock);
}

void SstWriterClose(SstStream Stream)
{
    struct _WriterCloseMsg Msg;
    struct timeval CloseTime, Diff;
    Msg.FinalTimestep = Stream->LastProvidedTimestep;
    sendOneToEachReaderRank(Stream, Stream->CPInfo->WriterCloseFormat, &Msg,
                            &Msg.RS_Stream);

    KillZeroRefTimesteps(Stream);

    /* wait until all queued data is sent */
    PTHREAD_MUTEX_LOCK(&Stream->DataLock);
    while (Stream->QueuedTimesteps)
    {
        CP_verbose(Stream,
                   "Waiting for timesteps to be released in WriterClose\n");
        if (Stream->Verbose)
        {
            CPTimestepList List = Stream->QueuedTimesteps;
            char *StringList = malloc(1);
            StringList[0] = 0;
            while (List)
            {
                char tmp[20];
                sprintf(tmp, "%ld ", List->Timestep);
                StringList =
                    realloc(StringList, strlen(StringList) + strlen(tmp) + 1);
                strcat(StringList, tmp);
                List = List->Next;
            }
            CP_verbose(Stream, "The timesteps still queued are: %s\n",
                       StringList);
            free(StringList);
        }
        CP_verbose(Stream, "Reader Count is %d\n", Stream->ReaderCount);
        for (int i = 0; i < Stream->ReaderCount; i++)
        {
            CP_verbose(Stream, "Reader [%d] status is %s\n", i,
                       SSTStreamStatusStr[Stream->Readers[i]->ReaderStatus]);
        }
        /* NEED TO HANDLE FAILURE HERE */
        pthread_cond_wait(&Stream->DataCondition, &Stream->DataLock);
    }
    PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);

    gettimeofday(&CloseTime, NULL);
    timersub(&CloseTime, &Stream->ValidStartTime, &Diff);
    if (Stream->Stats)
        Stream->Stats->ValidTimeSecs = (double)Diff.tv_usec / 1e6 + Diff.tv_sec;

    CP_verbose(Stream, "All timesteps are released in WriterClose\n");

    /*
     *  Only rank 0 removes contact info, and only when everything is closed.
     */
    if (Stream->Rank == 0)
    {
        removeContactInfo(Stream);
    }
}

#ifdef NOTDEF
static FFSFormatList AddUniqueFormats(FFSFormatList List,
                                      FFSFormatList Candidates)
{
    while (Candidates)
    {
        FFSFormatList Tmp = List;
        int found = 0;
        while (Tmp)
        {
            if ((Tmp->FormatIDRepLen == Candidates->FormatIDRepLen) &&
                (memcmp(Tmp->FormatIDRep, Candidates->FormatIDRep,
                        Tmp->FormatIDRepLen) == 0))
            {
                found++;
                break;
            }
            Tmp = Tmp->Next;
        }
        if (!found)
        {
            FFSFormatList New = malloc(sizeof(*New));
            memset(New, 0, sizeof(*New));
            New->FormatServerRep = malloc(Candidates->FormatServerRepLen);
            memcpy(New->FormatServerRep, Candidates->FormatServerRep,
                   Candidates->FormatServerRepLen);
            New->FormatServerRepLen = Candidates->FormatServerRepLen;
            New->FormatIDRep = malloc(Candidates->FormatIDRepLen);
            memcpy(New->FormatIDRep, Candidates->FormatIDRep,
                   Candidates->FormatIDRepLen);
            New->FormatIDRepLen = Candidates->FormatIDRepLen;
            New->Next = List;
            List = New;
        }
        Candidates = Candidates->Next;
    }
    return List;
}
#endif

static FFSFormatList AddUniqueFormats(FFSFormatList List,
                                      FFSFormatList Candidates, int copy)
{
    FFSFormatList Tmp = List;
    FFSFormatList Ret = List;

    // If nothing to add, return original
    if (!Candidates)
        return Ret;

    // Add tail of candidates list first
    Ret = AddUniqueFormats(List, Candidates->Next, copy);

    while (Tmp)
    {
        if ((Tmp->FormatIDRepLen == Candidates->FormatIDRepLen) &&
            (memcmp(Tmp->FormatIDRep, Candidates->FormatIDRep,
                    Tmp->FormatIDRepLen) == 0))
        {
            // Identical format already in List, don't add this one
            return Ret;
        }
        Tmp = Tmp->Next;
    }
    // New format not in list, add him to head and return.
    if (copy)
    {
        // Copy top Candidates entry before return
        FFSFormatList Tmp = malloc(sizeof(*Tmp));
        memset(Tmp, 0, sizeof(*Tmp));
        Tmp->FormatServerRep = malloc(Candidates->FormatServerRepLen);
        memcpy(Tmp->FormatServerRep, Candidates->FormatServerRep,
               Candidates->FormatServerRepLen);
        Tmp->FormatServerRepLen = Candidates->FormatServerRepLen;
        Tmp->FormatIDRep = malloc(Candidates->FormatIDRepLen);
        memcpy(Tmp->FormatIDRep, Candidates->FormatIDRep,
               Candidates->FormatIDRepLen);
        Tmp->FormatIDRepLen = Candidates->FormatIDRepLen;
        Candidates = Tmp;
    }
    Candidates->Next = Ret;
    return Candidates;
}

static void DoStreamDiscard(SstStream Stream)
{
    /* No readers, discard earliest timestep in queue (at the end of TS
     * list) */
    if (!Stream->QueuedTimesteps)
    {
        fprintf(stderr, "Expected at least one timestep to dequeue!\n");
    }
    if (Stream->QueuedTimesteps->Next == NULL)
    {
        CPTimestepList Entry = Stream->QueuedTimesteps;
        if (Stream->QueuedTimesteps->ReferenceCount != 0)
        {
            fprintf(stderr,
                    "\n\n\nREFERENCE COUNT CONDITION VIOLATION\n\n\n\n");
        }
        CP_verbose(Stream,
                   "Discarding timestep %d because of queue condition\n",
                   Entry->Timestep);
        Stream->QueuedTimesteps = NULL;
        Entry->FreeTimestep(Entry->FreeClientData);
        free(Entry->Msg);
        free(Entry->DataBlockToFree);
        free(Entry);
        Stream->QueuedTimestepCount--;
    }
    else
    {
        CPTimestepList Last = Stream->QueuedTimesteps;
        while (Last->Next->Next != NULL)
        {
            Last = Last->Next;
        }
        CPTimestepList Entry = Last->Next;
        if (Entry->ReferenceCount != 0)
        {
            fprintf(stderr,
                    "\n\n\nREFERENCE COUNT CONDITION VIOLATION\n\n\n\n");
        }
        CP_verbose(Stream,
                   "Discarding timestep %d because of queue condition\n",
                   Entry->Timestep);
        Last->Next = NULL;
        Entry->FreeTimestep(Entry->FreeClientData);
        free(Entry->MetadataArray);
        free(Entry->DP_TimestepInfo);
        free(Entry->DataBlockToFree);
        free(Entry->Msg);
        free(Entry);
        Stream->QueuedTimestepCount--;
    }
}

static void DoWriterSideGlobalOp(SstStream Stream, int *DiscardIncomingTimestep)
{
    int SendBlockSize = Stream->ReaderCount + 2;
    int *SendBlock = malloc(sizeof(SendBlock[0]) * SendBlockSize);
    int *RecvBlock =
        malloc(sizeof(SendBlock[0]) * SendBlockSize * Stream->CohortSize);

    PTHREAD_MUTEX_LOCK(&Stream->DataLock);
    CP_verbose(Stream, "Initiating Writer Side global operation\n");
    Stream->GlobalOpRequired = 0; //  reset if set
    RequestQueue ArrivingReader = Stream->ReadRequestQueue;

    /* build local info item -- contents:
     *  - Reader_requests_join (integer, contains count, only rank 0 should ever
     *set)
     *  - Current event queue length for this rank
     *	- Reader_status (integer, repeated ReaderCount times)
     */
    SendBlock[0] = 0;
    while (ArrivingReader)
    {
        SendBlock[0]++;
        ArrivingReader = ArrivingReader->Next;
    }
    if (Stream->QueueLimit &&
        ((Stream->QueuedTimestepCount + 1) > Stream->QueueLimit))
    {
        CP_verbose(Stream,
                   "Writer will be over queue limit, count %d, limit %d\n",
                   Stream->QueuedTimestepCount, Stream->QueueLimit);

        SendBlock[1] =
            1; /* this rank will be over queue limit with new timestep */
    }
    else
    {
        SendBlock[1] = 0; /* No problems here */
    }
    for (int i = 0; i < Stream->ReaderCount; i++)
    {
        SendBlock[i + 2] = Stream->Readers[i]->ReaderStatus;
    }
    MPI_Allgather(SendBlock, SendBlockSize, MPI_INT, RecvBlock, SendBlockSize,
                  MPI_INT, Stream->mpiComm);

    /*
     *  Handle incoming data in a deterministic way.  First handle any changes
     * in state for existing clients.
     */
    int ActiveReaderCount = 0;
    for (int i = 0; i < Stream->ReaderCount; i++)
    {

        if (Stream->Readers[i]->ReaderStatus == Established)
        {
            /*
             * if we think a reader is OK, see if anyone has had a failure with
             * it
             */
            for (int j = 0; j < Stream->CohortSize; j++)
            {
                if (RecvBlock[j * SendBlockSize + i + 2] == PeerFailed)
                {
                    Stream->Readers[i]->ReaderStatus = PeerFailed;
                }
            }
        }
        if (Stream->Readers[i]->ReaderStatus == PeerFailed)
        {
            // If we see PeerFailed now, everyone does, move to Fail
            CP_verbose(Stream, "Reader %d now determined to have failed, "
                               "dereferencing the timesteps it was sent, %d to "
                               "%d\n",
                       i, Stream->Readers[i]->StartingTimestep,
                       Stream->Readers[i]->LastSentTimestep);
            PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
            CP_PeerFailCloseWSReader(Stream->Readers[i], Closed);
            PTHREAD_MUTEX_LOCK(&Stream->DataLock);
        }
        if (Stream->Readers[i]->ReaderStatus == Established)
            ActiveReaderCount++;
    }

    int OverLimit = 0;
    for (int i = 0; i < Stream->CohortSize; i++)
    {
        if (RecvBlock[i * SendBlockSize + 1] > 0)
            OverLimit++;
    }
    /* we've got all the state we need to know, release data lock */
    PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);

    /*
     * before we add any new readers, see if we are going to need to
     * discard anything in the queue to stay in the queue limit
     */

    if (OverLimit)
    {
        if (Stream->QueueFullPolicy == SstQueueFullDiscard)
        {
            /* discard things */
            if (ActiveReaderCount == 0)
            {
                PTHREAD_MUTEX_LOCK(&Stream->DataLock);
                /*
                 * Have to double-check here.  While in general, if everyone
                 * had zero readers at the start, everyone should have the
                 * same set of timesteps queued and everyone should be doing
                 * a dequeue here.  However, we might have just gone to zero
                 * active (because of failed or exiting readers), and some
                 * timesteps might just have been released.  (Note: Assuming
                 * that if having gone to zero readers will have dequeued
                 * timesteps that had reference counts.  So generally we
                 * must be dequeueing things with a zero reference count
                 * here.)  That is an assumption that should not be
                 * violated.
                 */
                if ((Stream->QueuedTimestepCount + 1) >= Stream->QueueLimit)
                {
                    CP_verbose(Stream, "Writer doing discard for overlimit\n");
                    DoStreamDiscard(Stream);
                    OverLimit = 0; /* handled */
                }
                PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
            }
        }
    }

    /*
     *  Then handle possible incoming connection requests.  (Only rank 0 has
     * valid info, so only look to RecvBlock[0].)
     */
    for (int i = 0; i < RecvBlock[0]; i++)
    {
        WS_ReaderInfo reader;
        CP_verbose(Stream,
                   "Writer side Global operation accepting incoming reader\n");
        reader = WriterParticipateInReaderOpen(Stream);
        if (!reader)
        {
            CP_error(Stream, "Potential reader registration failed\n");
            break;
        }
        waitForReaderResponseAndSendQueued(reader);
    }

    /*
     *  Lastly, we'll decide what to do with the current provided timestep,
     *  (if it was not discarded before we added new readers).
     *
     *  If any writer rank is over the queuelimit, we must discard or block
     * decision points:
     If mode is block on queue limit:
         If there are no currently active readers, every node must block on
     reader arrival.
         else if there are active readers, each rank can block on queue size
     decreasing (as it will with progress)
     if mode is discard:
          If there are no currently active readers, we can discard the oldest
     item in the queue
          else if there are active readers, we must discard the currently
     arriving timestep.
     */
    if (OverLimit)
    {
        if (Stream->QueueFullPolicy == SstQueueFullBlock)
        {
            if (ActiveReaderCount == 0)
            {
                while (ActiveReaderCount == 0)
                {
                    WS_ReaderInfo reader;
                    if (Stream->Rank == 0)
                    {
                        PTHREAD_MUTEX_LOCK(&Stream->DataLock);
                        while (Stream->ReadRequestQueue == NULL)
                        {
                            pthread_cond_wait(&Stream->DataCondition,
                                              &Stream->DataLock);
                        }
                        PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
                    }
                    MPI_Barrier(Stream->mpiComm);
                    CP_verbose(Stream, "Writer side Global operation blocking "
                                       "for incoming reader\n");
                    reader = WriterParticipateInReaderOpen(Stream);
                    if (!reader)
                    {
                        CP_error(Stream,
                                 "Potential reader registration failed\n");
                        break;
                    }
                    waitForReaderResponseAndSendQueued(reader);
                    ActiveReaderCount++;
                }
            }
            else
            {
                /* wait until enough queued data is sent */
                PTHREAD_MUTEX_LOCK(&Stream->DataLock);
                while (Stream->QueueLimit &&
                       (Stream->QueuedTimestepCount > Stream->QueueLimit))
                {
                    CP_verbose(
                        Stream,
                        "Waiting for timesteps to be released in GlobalOp\n");
                    CP_verbose(Stream,
                               "The first timestep still queued is %d\n",
                               Stream->QueuedTimesteps->Timestep);
                    /* NEED TO HANDLE FAILURE HERE */
                    pthread_cond_wait(&Stream->DataCondition,
                                      &Stream->DataLock);
                }
                PTHREAD_MUTEX_UNLOCK(&Stream->DataLock);
            }
        }
        else
        {
            /* discard things */
            if (ActiveReaderCount == 0)
            {
                /* this should have been handled above */
                CP_verbose(Stream, "Finding a late need to discard when Active "
                                   "Readers is zero, shouldn't happen!!\n\n");
            }
            else
            {
                *DiscardIncomingTimestep = 1;
            }
        }
    }
    free(SendBlock);
    free(RecvBlock);
    CP_verbose(Stream, "Finished with Writer side Global operations\n");
}

extern void SstInternalProvideTimestep(
    SstStream Stream, SstData LocalMetadata, SstData Data, long Timestep,
    FFSFormatList Formats, DataFreeFunc FreeTimestep, void *FreeClientData,
    SstData AttributeData, DataFreeFunc FreeAttributeData,
    void *FreeAttributelientData)
{
    void *data_block;
    MetadataPlusDPInfo *pointers;
    struct _TimestepMetadataMsg *Msg = malloc(sizeof(*Msg));
    void *DP_TimestepInfo = NULL;
    struct _MetadataPlusDPInfo Md;
    CPTimestepList Entry = calloc(1, sizeof(struct _CPTimestepEntry));
    FFSFormatList XmitFormats = NULL;
    int GlobalOpRequested = 0;
    int GlobalOpRequired = 0;

    memset(Msg, 0, sizeof(*Msg));
    PTHREAD_MUTEX_LOCK(&Stream->DataLock);
{'fraction_non_alphanumeric': 0.09800515816621927, 'fraction_numerical': 0.00276856676667193, 'mean_word_length': 1.9862626135613468, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 923, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 85, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4879191', 'n_tokens_mistral': 27552, 'n_tokens_neox': 26431, 'n_words': 6142}