<commit_before>/**
 *******************************************************************************
 * NodeManager.cpp                                                             *
 *                                                                             *
 * Node Manager, performs following task(s):                                   *
 *   - Initializes runtime environment                                         *
 *   - Create & run kv store, membership process                               *
 *   - Clean up created processes                                              *
 *******************************************************************************
 */

/*
 *******************************************************************************
 *  Headers                                                                    *
 *******************************************************************************
 */
#include "stdinclude.h"
#include "NodeManager.h"

#include <sys/types.h>
#include <sys/wait.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <signal.h>
#include <syslog.h>
#include <pthread.h>
#include <errno.h>
#include <string.h>

#include "memberlist.h"
/*
 *******************************************************************************
 *  Declaractions                                                              *
 *******************************************************************************
 */

/*
 *******************************************************************************
 *  Type definitions                                                           *
 *******************************************************************************
 */

NodeManager thenode;
/*
 *******************************************************************************
 *  Functions                                                                  *
 *******************************************************************************
 */
NodeManager::NodeManager() :
    m_pid_membershipproc(0),
    m_pid_storeproc(0),
    m_b_running(false),
    m_addr(nullptr),
    m_shmid(-1),
    m_shm(nullptr)
{
}

NodeManager::~NodeManager()
{
}

int NodeManager::initialize()
{
    key_t ipckey = GetConfigPortal()->get_ipckey();
    long ring_size = GetConfigPortal()->get_ringsize();
    // The struct is 4 byte alignment, there's one additional MemberEntry
    size_t size = sizeof(struct Membership) + 
                  sizeof(struct MemberEntry) * ring_size; 

#define SHM_MODE  0600  /* User read/write */
    m_shmid = shmget(ipckey, size, IPC_CREAT|IPC_EXCL|SHM_MODE);
    if (m_shmid == -1) {
        syslog(LOG_ERR, "Create shared memory failed, errno=%d:%s\n", 
               errno, 
               strerror(errno));
        return -1;
    }
   
    m_addr = shmat(m_shmid, nullptr, 0);
    if (m_addr == (void*)-1) {
        syslog(LOG_ERR, "Shared memory attach failed, errno=%d:%s\n", 
               errno, 
               strerror(errno));
        return -1;
    }

    long *magic = (long*)m_addr;
    *magic = PL_SHM_MAGIC;

    if (!m_shm.attach(m_addr)) {
        syslog(LOG_ERR, "Member list attach failed\n");
        cleanup();
        return -1;
    }

    m_shm.set_ipckey(ipckey);
    m_shm.set_ring_size(ring_size);

    /* Initialize lock */
    if (init_lock()!=0) {
        cleanup();
        return -1;
    }

    m_shm.set_running_status(PL_RUNNING_ST_RUN);

    return 0;
}

int NodeManager::init_lock()
{
    pthread_mutex_t * plock = m_shm.get_mutex();
    if (plock == nullptr) {
        syslog(LOG_ERR, "Mutex address null\n");
        return -1;
    }

    pthread_mutexattr_t attr;
    if (pthread_mutexattr_init(&attr) != 0) {
        syslog(LOG_ERR, "Mutex attribute init failed, errno=%d:%s\n",
                        errno,
                        strerror(errno));
        return -1;
    }
    if (pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE) != 0) {
        syslog(LOG_ERR, "Mutex attribute set PTHREAD_MUTEX_RECURSIVE failed, errno=%d:%s\n",
                        errno,
                        strerror(errno));
        return -1;
    }
    if (pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED) != 0) {
        syslog(LOG_ERR, "Mutex attribute set PTHREAD_MUTEX_ROBUST failed, errno=%d:%s\n",
                        errno,
                        strerror(errno));
        return -1;
    }
    if (pthread_mutexattr_setrobust(&attr, PTHREAD_MUTEX_ROBUST) != 0) {
        syslog(LOG_ERR, "Mutex attribute set PTHREAD_MUTEX_ROBUST failed, errno=%d:%s\n",
                        errno,
                        strerror(errno));
        return -1;
    }

    return 0;
}

int NodeManager::startup()
{
    if ((start_membership() != 0) || 
        (start_store() != 0)) {
        getlog()->sendlog(FATAL, "Failed to startup process\n");
        terminate();
        return -1;
    }

    m_b_running = true;
    pid_t pid = 0;
    int status;
    while ((pid=wait(&status)) > 0) {
        if (m_b_running) {
            int rc = 0;
            if (pid == m_pid_membershipproc) {
                rc = start_membership();
            }
            else if (pid == m_pid_storeproc) {
                rc = start_store();
            }
            if (rc != 0) {<commit_msg>Update NodeManager.cpp<commit_after>/**
 *******************************************************************************
 * NodeManager.cpp                                                             *
 *                                                                             *
 * Node Manager, performs following task(s):                                   *
 *   - Initializes runtime environment                                         *
 *   - Create & run kv store, membership process                               *
 *   - Clean up created processes                                              *
 *******************************************************************************
 */

/*
 *******************************************************************************
 *  Headers                                                                    *
 *******************************************************************************
 */
#include "stdinclude.h"
#include "NodeManager.h"

#include <sys/types.h>
#include <sys/wait.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <signal.h>
#include <syslog.h>
#include <pthread.h>
#include <errno.h>
#include <string.h>

#include "memberlist.h"
/*
 *******************************************************************************
 *  Declaractions                                                              *
 *******************************************************************************
 */

/*
 *******************************************************************************
 *  Type definitions                                                           *
 *******************************************************************************
 */

NodeManager thenode;
/*
 *******************************************************************************
 *  Functions                                                                  *
 *******************************************************************************
 */
NodeManager::NodeManager() :
    m_pid_membershipproc(0),
    m_pid_storeproc(0),
    m_b_running(false),
    m_addr(nullptr),
    m_shmid(-1),
    m_shm(nullptr)
{
}

NodeManager::~NodeManager()
{
}

int NodeManager::initialize()
{
    key_t ipckey = GetConfigPortal()->get_ipckey();
    long ring_size = GetConfigPortal()->get_ringsize();
    // The struct is 4 byte alignment, there's one additional MemberEntry
    size_t size = sizeof(struct Membership) + 
                  sizeof(struct MemberEntry) * ring_size; 

#define SHM_MODE  0600  /* User read/write */
    m_shmid = shmget(ipckey, size, IPC_CREAT|IPC_EXCL|SHM_MODE);
    if (m_shmid == -1) {
        syslog(LOG_ERR, "Create shared memory failed, errno=%d:%s\n", 
               errno, 
               strerror(errno));
        return -1;
    }
   
    m_addr = shmat(m_shmid, nullptr, 0);
    if (m_addr == (void*)-1) {
        syslog(LOG_ERR, "Shared memory attach failed, errno=%d:%s\n", 
               errno, 
               strerror(errno));
        return -1;
    }

    long *magic = (long*)m_addr;
    *magic = PL_SHM_MAGIC;

    if (!m_shm.attach(m_addr)) {
        syslog(LOG_ERR, "Member list attach failed\n");
        cleanup();
        return -1;
    }

    m_shm.set_ipckey(ipckey);
    m_shm.set_ring_size(ring_size);

    /* Initialize lock */
    if (init_lock()!=0) {
        cleanup();
        return -1;
    }

    m_shm.set_running_status(PL_RUNNING_ST_RUN);

    return 0;
}

int NodeManager::init_lock()
{
    pthread_mutex_t * plock = m_shm.get_mutex();
    if (plock == nullptr) {
        syslog(LOG_ERR, "Mutex address null\n");
        return -1;
    }

    pthread_mutexattr_t attr;
    if (pthread_mutexattr_init(&attr) != 0) {
        syslog(LOG_ERR, "Mutex attribute init failed, errno=%d:%s\n",
                        errno,
                        strerror(errno));
        return -1;
    }
    if (pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE) != 0) {
        syslog(LOG_ERR, "Mutex attribute set PTHREAD_MUTEX_RECURSIVE failed, errno=%d:%s\n",
                        errno,
                        strerror(errno));
        return -1;
    }
    if (pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED) != 0) {
        syslog(LOG_ERR, "Mutex attribute set PTHREAD_MUTEX_ROBUST failed, errno=%d:%s\n",
                        errno,
                        strerror(errno));
        return -1;
    }
    if (pthread_mutexattr_setrobust(&attr, PTHREAD_MUTEX_ROBUST) != 0) {
        syslog(LOG_ERR, "Mutex attribute set PTHREAD_MUTEX_ROBUST failed, errno=%d:%s\n",
                        errno,
                        strerror(errno));
        return -1;
    }
    if (pthread_mutexattr_setprotocol(&attr, PTHREAD_PRIO_INHERIT) != 0) {
        syslog(LOG_ERR, "Mutex attribute set PTHREAD_PRIO_INHERIT failed, errno=%d:%s\n",
                        errno,
                        strerror(errno));
        return -1;
    }

    return 0;
}

int NodeManager::startup()
{
    if ((start_membership() != 0) || 
        (start_store() != 0)) {
        getlog()->sendlog(FATAL, "Failed to startup process\n");
        terminate();
        return -1;
    }

    m_b_running = true;
    pid_t pid = 0;
    int status;
    while ((pid=wait(&status)) > 0) {
        if (m_b_running) {
            int rc = 0;
            if (pid == m_pid_membershipproc) {
                rc = start_membership();
            }
            else if (pid == m_pid_storeproc) {
                rc = start_store();
            }
            if (rc != 0) {
{'fraction_non_alphanumeric': 0.24744850621636666, 'fraction_numerical': 0.006494711449248469, 'mean_word_length': 1.9108830677828788, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 34, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4397882', 'n_tokens_mistral': 3055, 'n_tokens_neox': 2870, 'n_words': 556}