Commit b5f8c324 authored by Greg Williams's avatar Greg Williams
Browse files

Added aggregate PUT/write bandwidth to overlapped IO test to assess overall... 

Added aggregate PUT/write bandwidth to overlapped IO test to assess overall throughput for analysis of overlapped sessions.
Fixed application of log level priority.
Bumbed max sessions from 6 to 20 in order to allow more overlapped sessions and greater throughput on powerful servers.
parent cd59b90a

Makefile

+1 −1
Original line number Diff line number Diff line
@@ -88,7 +88,7 @@ clean: 
$(OUT_DIR)/socket99.o: $(SOCKET99)/socket99.c $(SOCKET99)/socket99.h

	$(CC) -c -o $@ $< $(CFLAGS) -I$(SOCKET99)

$(OUT_DIR)/protobuf-c.o: $(PROTOBUFC)/protobuf-c/protobuf-c.c $(PROTOBUFC)/protobuf-c/protobuf-c.h

	$(CC) -c -o $@ $< -std=c99 -fPIC -g -Wall $(OPTIMIZE) -Wno-unused-parameter -I$(PROTOBUFC)

	$(CC) -c -o $@ $< -std=c99 -fPIC -g -Wall -Wno-unused-parameter $(OPTIMIZE) -I$(PROTOBUFC)

$(OUT_DIR)/kinetic_allocator.o: $(LIB_DIR)/kinetic_allocator.c $(LIB_DEPS)

	$(CC) -c -o $@ $< $(CFLAGS) $(LIB_INCS)

$(OUT_DIR)/kinetic_nbo.o: $(LIB_DIR)/kinetic_nbo.c $(LIB_DEPS)

config/create_mocks.rb

0 → 100644
+10 −0
Original line number Diff line number Diff line 
require 'cmock'



HERE = File.expand_path(File.join(File.dirname(__FILE__), '../'))

mock_out = File.join(HERE, 'build/test/mocks')

cmock = CMock.new(plugins: [:ignore, :return_thru_ptr], mock_path: mock_out)

headers_to_mock = Dir['src/lib/*.h']

puts "\nCreating mocks from header files...\n-----------------------------------"

cmock.setup_mocks(headers_to_mock)

puts "\nGenerated Mocks:\n----------------"

Dir["#{mock_out}/mock_*.h"].each{|mock| puts mock}

src/lib/kinetic_logger.c

+111 −84
Original line number Diff line number Diff line
@@ -25,103 +25,43 @@ 
#include <pthread.h>

#include <sys/time.h>



// #define USE_GENERIC_LOGGER 1

// #define USE_GENERIC_LOGGER 1 (not ready yet!)



#define KINETIC_LOGGER_BUFFER_STR_MAX_LEN 256

#define KINETIC_LOGGER_BUFFER_SIZE (0x1 << 12)

#define KINETIC_LOGGER_FLUSH_INTERVAL_SEC 180

#define KINETIC_LOGGER_SLEEP_TIME_SEC 10

#define KINETIC_LOGGER_BUFFER_FLUSH_SIZE (0.8 * KINETIC_LOGGER_BUFFER_SIZE)

#define KINETIC_LOGGER_FLUSH_THREAD_ENABLED false



STATIC int KineticLogLevel = -1;

STATIC FILE* KineticLoggerHandle = NULL;

STATIC pthread_mutex_t KineticLoggerBufferMutex = PTHREAD_MUTEX_INITIALIZER;

STATIC char KineticLoggerBuffer[KINETIC_LOGGER_BUFFER_SIZE][KINETIC_LOGGER_BUFFER_STR_MAX_LEN];

STATIC int KineticLoggerBufferSize = 0;

STATIC bool KineticLoggerForceFlush = false;



static inline void KineticLogger_BufferLock(void)

{

    pthread_mutex_lock(&KineticLoggerBufferMutex);

}



static inline void KineticLogger_BufferUnlock()

{

    pthread_mutex_unlock(&KineticLoggerBufferMutex);

}



static void KineticLogger_FlushBuffer(void)

{

    if (KineticLoggerHandle == NULL) {

        return;

    }

    for (int i = 0; i < KineticLoggerBufferSize; i++) {

        fprintf(KineticLoggerHandle, "%s", KineticLoggerBuffer[i]);

    }

    fflush(KineticLoggerHandle);

    KineticLoggerBufferSize = 0;

}



static inline char* KineticLogger_GetBuffer()

{

    KineticLogger_BufferLock();

    if (KineticLoggerBufferSize >= KINETIC_LOGGER_BUFFER_SIZE) {

        KineticLogger_FlushBuffer();

    }



    // allocate buffer

    KineticLoggerBufferSize++;

    return KineticLoggerBuffer[KineticLoggerBufferSize-1];

}



static inline void KineticLogger_FinishBuffer(void)

{

    if (KineticLoggerForceFlush) {

        KineticLogger_FlushBuffer();

    }

    KineticLogger_BufferUnlock();

}



#if KINETIC_LOGGER_FLUSH_THREAD_ENABLED

static void* KineticLogger_FlushThread(void* arg)

{

    (void)arg;

    struct timeval tv;

    time_t lasttime;

    time_t curtime;

STATIC bool KineticLoggerForceFlush = false;

#else

STATIC bool KineticLoggerForceFlush = true;

#endif



    gettimeofday(&tv, NULL);



    lasttime = tv.tv_sec;

//------------------------------------------------------------------------------

// Private Method Declarations



    for(;;) {

        sleep(KINETIC_LOGGER_SLEEP_TIME_SEC);

        gettimeofday(&tv, NULL);

        curtime = tv.tv_sec;

        if ((curtime - lasttime) >= KINETIC_LOGGER_FLUSH_INTERVAL_SEC) {

            KineticLogger_FlushBuffer();

            lasttime = curtime;

        }

        else {

            KineticLogger_BufferLock();

            if (KineticLoggerBufferSize >= KINETIC_LOGGER_BUFFER_FLUSH_SIZE) {

                KineticLogger_FlushBuffer();

            }

            KineticLogger_BufferUnlock();

        }

    }

static inline void KineticLogger_BufferLock(void);

static inline void KineticLogger_BufferUnlock(void);

static void KineticLogger_FlushBuffer(void);

static inline char* KineticLogger_GetBuffer(void);

static inline void KineticLogger_FinishBuffer(void);

#if KINETIC_LOGGER_FLUSH_THREAD_ENABLED

static void* KineticLogger_FlushThread(void* arg);

static void KineticLogger_InitFlushThread(void);

#endif



    return NULL;

}



static void KineticLogger_InitFlushThread(void)

{

    pthread_t thr;

    pthread_create(&thr, NULL, KineticLogger_FlushThread, NULL);

    KineticLogger_Log(0, "Flush thread is created.\n");

}

#endif

//------------------------------------------------------------------------------

// Public Method Definitions



void KineticLogger_Init(const char* log_file, int log_level)

{
@@ -136,11 +76,11 @@ void KineticLogger_Init(const char* log_file, int log_level) 
        KineticLogLevel = log_level;

        

        if (strncmp(log_file, "stdout", 4) == 0 || strncmp(log_file, "STDOUT", 4) == 0) {

            printf("\nLogging kinetic-c output to console (stdout)\n");

            printf("\nLogging kinetic-c output to console (stdout) w/ log_level=%d\n", KineticLogLevel);

            KineticLoggerHandle = stdout;

        }

        else {

            printf("\nLogging kinetic-c output to %s\n", log_file);

            printf("\nLogging kinetic-c output to %s w/ log_level=%d\n", log_file, KineticLogLevel);

            KineticLoggerHandle = fopen(log_file, "a+");

            assert(KineticLoggerHandle != NULL);

        }
@@ -148,6 +88,7 @@ void KineticLogger_Init(const char* log_file, int log_level) 
        // Create thread to periodically flush the log

        #if KINETIC_LOGGER_FLUSH_THREAD_ENABLED

        KineticLogger_InitFlushThread();

        KineticLoggerForceFlush = false;

        #endif

    }

}
@@ -165,7 +106,7 @@ void KineticLogger_Close(void) 


void KineticLogger_Log(int log_level, const char* message)

{

    if (message == NULL || log_level < KineticLogLevel || KineticLogLevel < 0) {

    if (message == NULL || log_level > KineticLogLevel || KineticLogLevel < 0) {

        return;

    }
@@ -173,12 +114,11 @@ void KineticLogger_Log(int log_level, const char* message) 
    buffer = KineticLogger_GetBuffer();

    snprintf(buffer, KINETIC_LOGGER_BUFFER_STR_MAX_LEN, "%s\n", message);

    KineticLogger_FinishBuffer();

    KineticLogger_FlushBuffer();

}



void KineticLogger_LogPrintf(int log_level, const char* format, ...)

{

    if (format == NULL || log_level < KineticLogLevel || KineticLogLevel < 0) {

    if (format == NULL || log_level > KineticLogLevel || KineticLogLevel < 0) {

        return;

    }
@@ -192,7 +132,6 @@ void KineticLogger_LogPrintf(int log_level, const char* format, ...) 


    strcat(buffer, "\n");

    KineticLogger_FinishBuffer();

    KineticLogger_FlushBuffer();

}



void KineticLogger_LogLocation(const char* filename, int line, const char* message)
@@ -825,3 +764,91 @@ void KineticLogger_LogByteBuffer(int log_level, const char* title, ByteBuffer bu 
    ByteArray array = {.data = buffer.array.data, .len = buffer.bytesUsed};

    KineticLogger_LogByteArray(log_level, title, array);

}





//------------------------------------------------------------------------------

// Private Method Definitions



static inline void KineticLogger_BufferLock(void)

{

    pthread_mutex_lock(&KineticLoggerBufferMutex);

}



static inline void KineticLogger_BufferUnlock()

{

    pthread_mutex_unlock(&KineticLoggerBufferMutex);

}



static void KineticLogger_FlushBuffer(void)

{

    if (KineticLoggerHandle == NULL) {

        return;

    }

    for (int i = 0; i < KineticLoggerBufferSize; i++) {

        fprintf(KineticLoggerHandle, "%s", KineticLoggerBuffer[i]);

    }

    fflush(KineticLoggerHandle);

    KineticLoggerBufferSize = 0;

}



static inline char* KineticLogger_GetBuffer()

{

    KineticLogger_BufferLock();

    if (KineticLoggerBufferSize >= KINETIC_LOGGER_BUFFER_SIZE) {

        KineticLogger_FlushBuffer();

    }



    // allocate buffer

    KineticLoggerBufferSize++;

    return KineticLoggerBuffer[KineticLoggerBufferSize-1];

}



static inline void KineticLogger_FinishBuffer(void)

{

    if (KineticLoggerForceFlush) {

        KineticLogger_FlushBuffer();

    }

    KineticLogger_BufferUnlock();

}



#if KINETIC_LOGGER_FLUSH_THREAD_ENABLED



static void* KineticLogger_FlushThread(void* arg)

{

    (void)arg;

    struct timeval tv;

    time_t lasttime;

    time_t curtime;



    gettimeofday(&tv, NULL);



    lasttime = tv.tv_sec;



    for(;;) {

        sleep(KINETIC_LOGGER_SLEEP_TIME_SEC);

        gettimeofday(&tv, NULL);

        curtime = tv.tv_sec;

        if ((curtime - lasttime) >= KINETIC_LOGGER_FLUSH_INTERVAL_SEC) {

            KineticLogger_FlushBuffer();

            lasttime = curtime;

        }

        else {

            KineticLogger_BufferLock();

            if (KineticLoggerBufferSize >= KINETIC_LOGGER_BUFFER_FLUSH_SIZE) {

                KineticLogger_FlushBuffer();

            }

            KineticLogger_BufferUnlock();

        }

    }



    return NULL;

}



static void KineticLogger_InitFlushThread(void)

{

    pthread_t thr;

    pthread_create(&thr, NULL, KineticLogger_FlushThread, NULL);

    KineticLogger_Log(0, "Flush thread is created.\n");

}



#endif

src/lib/kinetic_logger.h

+1 −0
Original line number Diff line number Diff line
@@ -26,6 +26,7 @@ 
#include <stdarg.h>



#define KINETIC_LOGGER_DISABLED false

#define KINETIC_LOGGER_FLUSH_THREAD_ENABLED true

#define KINETIC_LOG_FILE "kinetic.log"



void KineticLogger_Init(const char* logFile, int log_level);

src/lib/kinetic_types_internal.h

+1 −1
Original line number Diff line number Diff line
@@ -29,7 +29,7 @@ 
#include <time.h>

#include <pthread.h>



#define KINETIC_SESSIONS_MAX (6)

#define KINETIC_SESSIONS_MAX (20)

#define KINETIC_PDUS_PER_SESSION_DEFAULT (2)

#define KINETIC_PDUS_PER_SESSION_MAX (10)

#define KINETIC_SOCKET_DESCRIPTOR_INVALID (-1)