<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 "adios2/toolkit/profiling/taustubs/taustubs.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);

static int locked = 0;
#ifdef MUTEX_DEBUG
#define PTHREAD_MUTEX_LOCK(lock)                                               \
    printf("Trying lock line %d\n", __LINE__);                                 \
    pthread_mutex_lock(lock);                                                  \
    locked++;                                                                  \
    printf("Got lock\n");
#define PTHREAD_MUTEX_UNLOCK(lock)                                             \
    printf("UNlocking line %d\n", __LINE__);                                   \
    locked--;                                                                  \
    pthread_mutex_unlock(lock);
#define SST_ASSERT_LOCKED() assert(locked)
#else
#define PTHREAD_MUTEX_LOCK(lock)                                               \
    {                                                                          \
        pthread_mutex_lock(lock);                                              \
        locked++;                                                              \
    }
#define PTHREAD_MUTEX_UNLOCK(lock)                                             \
    {                                                                          \
        locked--;                                                              \
        pthread_mutex_unlock(lock);                                            \
    }
#define SST_ASSERT_LOCKED() assert(locked)
#define SST_ASSERT_UNLOCKED() assert(unlocked)
#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;
    }
}

/*
RemoveQueueEntries:
        If the number of timesteps older than OldestCurrentReaderTimestep, mark
them as Expired Dequeue and free any timestep that is Expired, not Precious and
has reference count 0. if change SIGNAL

*/
static void RemoveQueueEntries(SstStream Stream)
{
    int AnythingRemoved = 0;
    CPTimestepList List = Stream->QueuedTimesteps;
    CPTimestepList Last = NULL;
    CPTimestepList Prev = NULL;

    CP_verbose(Stream, "In remove queue entries\n");
    while (List)
    {
        CPTimestepList Next = List->Next;
        int Freed = 0;
        if (List->Expired && (!List->PreciousTimestep) &&
            (List->ReferenceCount == 0))
        {
            CPTimestepList ItemToFree = List;
            Freed = 1;
            if (ItemToFree->DPRegistered)
            {
                Stream->DP_Interface->releaseTimestep(&Svcs, Stream->DP_Stream,
                                                      List->Timestep);
            }

            Stream->QueuedTimestepCount--;
            CP_verbose(Stream,
                       "Remove queue Entries removing Timestep %ld (exp %d, "
                       "Prec %d, Ref %d), Count now %d\n",
                       ItemToFree->Timestep, ItemToFree->Expired,
                       ItemToFree->PreciousTimestep, ItemToFree->ReferenceCount,
                       Stream->QueuedTimestepCount);
            ItemToFree->FreeTimestep(ItemToFree->FreeClientData);
            free(ItemToFree->Msg);
            //            free(ItemToFree->MetadataArray);
            //            free(ItemToFree->DP_TimestepInfo);
            free(ItemToFree->DataBlockToFree);
            free(ItemToFree);
            AnythingRemoved++;

            if (Last)
            {
                /* unlink item */
                Last->Next = Next;
            }
            else
            {
                Stream->QueuedTimesteps = Next;
            }
        }
        if (!Freed)
        {
            Last = List;
        }
        List = Next;
    }

    if (AnythingRemoved)
    {
        /* main thread might be waiting on timesteps going away */
        pthread_cond_signal(&Stream->DataCondition);
    }
}

static void ReleaseAndDiscardRemainingTimesteps(SstStream Stream)
{
    CPTimestepList List = Stream->QueuedTimesteps;

    while (List)
    {
        List->Expired = 1;
        List->PreciousTimestep = 0;
        List->ReferenceCount = 0;
        List = List->Next;
    }
    RemoveQueueEntries(Stream);
}

/*
Queue maintenance:    (ASSUME LOCKED)
        calculate smallest entry for CurrentTimestep in a reader.  Update that
as OldestCurrentReaderTimestep. If any timestep has zero ref count and is
registered with DP deregister that timestep with DP CallRemoveQueueEntries
*/
static void QueueMaintenance(SstStream Stream)
{
    SST_ASSERT_LOCKED();
    long SmallestLastReleasedTimestep = LONG_MAX;
    long ReserveCount = Stream->ConfigParams->ReserveQueueLimit;

    CP_verbose(Stream, "In queue maintenance\n");
    CPTimestepList List;
    for (int i = 0; i < Stream->ReaderCount; i++)
    {
        CP_verbose(Stream, "Reader %d status %s has last released %ld\n", i,
                   SSTStreamStatusStr[Stream->Readers[i]->ReaderStatus],
                   Stream->Readers[i]->LastReleasedTimestep);
        if (Stream->Readers[i]->ReaderStatus == Established)
        {
            if (Stream->Readers[i]->LastReleasedTimestep <
                SmallestLastReleasedTimestep)
                SmallestLastReleasedTimestep =
                    Stream->Readers[i]->LastReleasedTimestep;
        }
    }
    CP_verbose(Stream,
               "QueueMaintenance, smallest last released = %ld, count = %d\n",
               SmallestLastReleasedTimestep, Stream->QueuedTimestepCount);
    /* Count precious */<commit_msg>Use status<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 "adios2/toolkit/profiling/taustubs/taustubs.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);

static int locked = 0;
#ifdef MUTEX_DEBUG
#define PTHREAD_MUTEX_LOCK(lock)                                               \
    printf("Trying lock line %d\n", __LINE__);                                 \
    pthread_mutex_lock(lock);                                                  \
    locked++;                                                                  \
    printf("Got lock\n");
#define PTHREAD_MUTEX_UNLOCK(lock)                                             \
    printf("UNlocking line %d\n", __LINE__);                                   \
    locked--;                                                                  \
    pthread_mutex_unlock(lock);
#define SST_ASSERT_LOCKED() assert(locked)
#else
#define PTHREAD_MUTEX_LOCK(lock)                                               \
    {                                                                          \
        pthread_mutex_lock(lock);                                              \
        locked++;                                                              \
    }
#define PTHREAD_MUTEX_UNLOCK(lock)                                             \
    {                                                                          \
        locked--;                                                              \
        pthread_mutex_unlock(lock);                                            \
    }
#define SST_ASSERT_LOCKED() assert(locked)
#define SST_ASSERT_UNLOCKED() assert(unlocked)
#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;
    }
}

/*
RemoveQueueEntries:
        If the number of timesteps older than OldestCurrentReaderTimestep, mark
them as Expired Dequeue and free any timestep that is Expired, not Precious and
has reference count 0. if change SIGNAL

*/
static void RemoveQueueEntries(SstStream Stream)
{
    int AnythingRemoved = 0;
    CPTimestepList List = Stream->QueuedTimesteps;
    CPTimestepList Last = NULL;
    CPTimestepList Prev = NULL;

    CP_verbose(Stream, "In remove queue entries\n");
    while (List)
    {
        CPTimestepList Next = List->Next;
        int Freed = 0;
        if (List->Expired && (!List->PreciousTimestep) &&
            (List->ReferenceCount == 0))
        {
            CPTimestepList ItemToFree = List;
            Freed = 1;
            if (ItemToFree->DPRegistered)
            {
                Stream->DP_Interface->releaseTimestep(&Svcs, Stream->DP_Stream,
                                                      List->Timestep);
            }

            Stream->QueuedTimestepCount--;
            CP_verbose(Stream,
                       "Remove queue Entries removing Timestep %ld (exp %d, "
                       "Prec %d, Ref %d), Count now %d\n",
                       ItemToFree->Timestep, ItemToFree->Expired,
                       ItemToFree->PreciousTimestep, ItemToFree->ReferenceCount,
                       Stream->QueuedTimestepCount);
            ItemToFree->FreeTimestep(ItemToFree->FreeClientData);
            free(ItemToFree->Msg);
            //            free(ItemToFree->MetadataArray);
            //            free(ItemToFree->DP_TimestepInfo);
            free(ItemToFree->DataBlockToFree);
            free(ItemToFree);
            AnythingRemoved++;

            if (Last)
            {
                /* unlink item */
                Last->Next = Next;
            }
            else
            {
                Stream->QueuedTimesteps = Next;
            }
        }
        if (!Freed)
        {
            Last = List;
        }
        List = Next;
    }

    if (AnythingRemoved)
    {
        /* main thread might be waiting on timesteps going away */
        pthread_cond_signal(&Stream->DataCondition);
    }
}

static void ReleaseAndDiscardRemainingTimesteps(SstStream Stream)
{
    CPTimestepList List = Stream->QueuedTimesteps;

    while (List)
    {
        List->Expired = 1;
        List->PreciousTimestep = 0;
        List->ReferenceCount = 0;
        List = List->Next;
    }
    RemoveQueueEntries(Stream);
}

/*
Queue maintenance:    (ASSUME LOCKED)
        calculate smallest entry for CurrentTimestep in a reader.  Update that
as OldestCurrentReaderTimestep. If any timestep has zero ref count and is
registered with DP deregister that timestep with DP CallRemoveQueueEntries
*/
static void QueueMaintenance(SstStream Stream)
{
    SST_ASSERT_LOCKED();
    long SmallestLastReleasedTimestep = LONG_MAX;
    long ReserveCount;

    if (Stream->Status != Established)
        return;

    ReserveCount = Stream->ConfigParams->ReserveQueueLimit;
    CP_verbose(Stream, "In queue maintenance\n");
    CPTimestepList List;
    for (int i = 0; i < Stream->ReaderCount; i++)
    {
        CP_verbose(Stream, "Reader %d status %s has last released %ld\n", i,
                   SSTStreamStatusStr[Stream->Readers[i]->ReaderStatus],
                   Stream->Readers[i]->LastReleasedTimestep);
        if (Stream->Readers[i]->ReaderStatus == Established)
        {
            if (Stream->Readers[i]->LastReleasedTimestep <
                SmallestLastReleasedTimestep)
                SmallestLastReleasedTimestep =
                    Stream->Readers[i]->LastReleasedTimestep;
        }
    }
    CP_verbose(Stream,
               "QueueMaintenance, smallest last released = %ld, count = %d\n",
               SmallestLastReleasedTimestep, Stream->QueuedTimestepCount);
    /* Count precious */
{'fraction_non_alphanumeric': 0.09820491024551227, 'fraction_numerical': 0.0023001150057502875, 'mean_word_length': 1.8910089621277826, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 146, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 70, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3856835', 'n_tokens_mistral': 5704, 'n_tokens_neox': 5550, 'n_words': 1292}