<commit_before>#include "marquise.h"
#include "structs.h"
#include "frame.h"
#include "macros.h"
#include "../config.h"
#include "defer.h"

#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>
#include <zmq.h>
#include <stdio.h>
#include <glib.h>
#include <sys/mman.h>
#include "lz4/lz4.h"
#include "lz4/lz4hc.h"
#include <string.h>
#include <math.h>

static void debug_log( char *format, ...) {
        if( !getenv( "LIBMARQUISE_DEBUG" ) )
                return;

        va_list args;
        va_start( args, format );
        vfprintf( stderr, format, args );
        va_end( args );
}

static void *connect_upstream_socket( void *context, char *broker ) {
        debug_log( "Initializing upstream socket to %s\n", broker );
        // Set up the upstream REQ socket.
        void *upstream_connection = zmq_socket( context, ZMQ_REQ );

        fail_if( zmq_connect( upstream_connection, broker )
                   , zmq_close( upstream_connection ); return NULL;
                   , "zmq_connect: '%s'"
                   , strerror( errno ) );

        // We want to be explicit about the internal high water mark:
        // After this many poll periods have elapsed, zmq_sendmsg will start
        // returning EAGAIN and we will start deferring further messages to
        // disk to save memory.
        int one = 1;

        fail_if( zmq_setsockopt( upstream_connection, ZMQ_SNDHWM, &one, sizeof( one ) )
                   , zmq_close( upstream_connection );
                   , "zmq_setsockopt ZMQ_HWM: '%s'"
                   , strerror( errno ) );

        // Milliseconds
        int timeout = 10000;
        fail_if( zmq_setsockopt( upstream_connection
                                   , ZMQ_SNDTIMEO
                                   , &timeout
                                   , sizeof( timeout ) )
                   , zmq_close( upstream_connection ); return NULL;
                   , "zmq_setsockopt ZMQ_SNDTIMEO: '%s'"
                   , strerror( errno ) );

        // Milliseconds
        fail_if( zmq_setsockopt( upstream_connection
                                   , ZMQ_RCVTIMEO
                                   , &timeout
                                   , sizeof( timeout ) )
                   , zmq_close( upstream_connection ); return NULL;
                   , "zmq_setsockopt ZMQ_SNDTIMEO: '%s'"
                   , strerror( errno ) );

        return upstream_connection;
}

static frame *accumulate_databursts( gpointer zmq_message, gint *queue_length )
{
        static frame *accumulator;
        static size_t offset = 0;

        // Reset and return result
        if( !zmq_message && !queue_length ) {
                debug_log( "Resetting accumulator\n" );

                frame *ret = accumulator;
                accumulator = NULL;
                offset = 0;
                return ret;
        }

        if( *queue_length <= 0 ) {
                return NULL;
        }

        if( !accumulator ) {
                debug_log( "Initializing accumulator for queue length %u\n"
                         , *queue_length );
                accumulator = calloc( *queue_length, sizeof( frame ) );
                if( !accumulator ) {
                        *queue_length -= 1;
                        return NULL;
                }
        }

        size_t msg_size = zmq_msg_size( zmq_message );
        accumulator[offset].length = msg_size;
        accumulator[offset].data = malloc( msg_size );
        if( !accumulator[offset].data ) {
                *queue_length -= 1;
                return NULL;
        }
        memcpy( accumulator[offset].data, zmq_msg_data( zmq_message ), msg_size );
        zmq_msg_close( zmq_message );
        free( zmq_message );
        offset++;

        return NULL;
}

// This function either sends the message or defers it. There is no try.
static void try_send_upstream( data_burst *burst, queue_args *args ) {
        debug_log( "Attempting to send burst of len %d upstream\n"
                 , burst->length );

        int ret;

        retry_send:
        // This will timeout due to ZMQ_SNDTIMEO being set on the socket.
        ret = zmq_send( args->upstream_connection
                      , burst->data
                      , burst->length
                      , 0 );
        // I belive that this is reduntant due to ZMQ_SNDTIMEO. It is more me
        // being paranoid.
        if( ret == -1 && errno == EINTR )
                goto retry_send;
        debug_log( "Upstream zmq_send() returned %d\n", ret );

        fail_if( ret == -1
               , marquise_defer_to_file( args->deferral_file, burst );
                 free_databurst( burst );
                 return;
               , "sending message: '%s'", strerror( errno ) );

        // This will timeout also due to ZMQ_RCVTIMEO
        zmq_msg_t response;
        zmq_msg_init( &response );

        retry_recv:
        ret = zmq_msg_recv( &response, args->upstream_connection, 0 );
        if( ret == -1 && errno == EINTR )
                goto retry_recv;
        debug_log( "Upstream zmq_recv() returned %d\n", ret );

        // If a recv fails, we need to reset the connection or the state
        // machine will be out of step.
        fail_if( ret == -1
               , marquise_defer_to_file( args->deferral_file, burst );
                 free_databurst( burst );
                 int linger = 0;
                 zmq_setsockopt( args->upstream_connection
                               , ZMQ_LINGER
                               , &linger
                               , sizeof( linger ) );
                 zmq_close( args->upstream_connection );
                 args->upstream_connection = connect_upstream_socket(
                           args->upstream_context
                         , args->broker );
                 return;
               , "receiving ack: '%s'", strerror( errno ) );

        // The second failure case is that upstream sends back an error as the
        // ack
        fail_if( ret > 0
               , marquise_defer_to_file( args->deferral_file, burst );
                 free_databurst( burst );
                 return;
               , "recieved upstream error: '%.*s'"
               , ret
               , (char *) zmq_msg_data( &response ) );

        // This may in future contain something useful, like a hash of the
        // message that we can verify.<commit_msg>Do not retry in the case of an error returning all the way from a vault writer daemon<commit_after>#include "marquise.h"
#include "structs.h"
#include "frame.h"
#include "macros.h"
#include "../config.h"
#include "defer.h"

#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>
#include <zmq.h>
#include <stdio.h>
#include <glib.h>
#include <sys/mman.h>
#include "lz4/lz4.h"
#include "lz4/lz4hc.h"
#include <string.h>
#include <math.h>

static void debug_log( char *format, ...) {
        if( !getenv( "LIBMARQUISE_DEBUG" ) )
                return;

        va_list args;
        va_start( args, format );
        vfprintf( stderr, format, args );
        va_end( args );
}

static void *connect_upstream_socket( void *context, char *broker ) {
        debug_log( "Initializing upstream socket to %s\n", broker );
        // Set up the upstream REQ socket.
        void *upstream_connection = zmq_socket( context, ZMQ_REQ );

        fail_if( zmq_connect( upstream_connection, broker )
                   , zmq_close( upstream_connection ); return NULL;
                   , "zmq_connect: '%s'"
                   , strerror( errno ) );

        // We want to be explicit about the internal high water mark:
        // After this many poll periods have elapsed, zmq_sendmsg will start
        // returning EAGAIN and we will start deferring further messages to
        // disk to save memory.
        int one = 1;

        fail_if( zmq_setsockopt( upstream_connection, ZMQ_SNDHWM, &one, sizeof( one ) )
                   , zmq_close( upstream_connection );
                   , "zmq_setsockopt ZMQ_HWM: '%s'"
                   , strerror( errno ) );

        // Milliseconds
        int timeout = 10000;
        fail_if( zmq_setsockopt( upstream_connection
                                   , ZMQ_SNDTIMEO
                                   , &timeout
                                   , sizeof( timeout ) )
                   , zmq_close( upstream_connection ); return NULL;
                   , "zmq_setsockopt ZMQ_SNDTIMEO: '%s'"
                   , strerror( errno ) );

        // Milliseconds
        fail_if( zmq_setsockopt( upstream_connection
                                   , ZMQ_RCVTIMEO
                                   , &timeout
                                   , sizeof( timeout ) )
                   , zmq_close( upstream_connection ); return NULL;
                   , "zmq_setsockopt ZMQ_SNDTIMEO: '%s'"
                   , strerror( errno ) );

        return upstream_connection;
}

static frame *accumulate_databursts( gpointer zmq_message, gint *queue_length )
{
        static frame *accumulator;
        static size_t offset = 0;

        // Reset and return result
        if( !zmq_message && !queue_length ) {
                debug_log( "Resetting accumulator\n" );

                frame *ret = accumulator;
                accumulator = NULL;
                offset = 0;
                return ret;
        }

        if( *queue_length <= 0 ) {
                return NULL;
        }

        if( !accumulator ) {
                debug_log( "Initializing accumulator for queue length %u\n"
                         , *queue_length );
                accumulator = calloc( *queue_length, sizeof( frame ) );
                if( !accumulator ) {
                        *queue_length -= 1;
                        return NULL;
                }
        }

        size_t msg_size = zmq_msg_size( zmq_message );
        accumulator[offset].length = msg_size;
        accumulator[offset].data = malloc( msg_size );
        if( !accumulator[offset].data ) {
                *queue_length -= 1;
                return NULL;
        }
        memcpy( accumulator[offset].data, zmq_msg_data( zmq_message ), msg_size );
        zmq_msg_close( zmq_message );
        free( zmq_message );
        offset++;

        return NULL;
}

// This function either sends the message or defers it. There is no try.
static void try_send_upstream( data_burst *burst, queue_args *args ) {
        debug_log( "Attempting to send burst of len %d upstream\n"
                 , burst->length );

        int ret;

        retry_send:
        // This will timeout due to ZMQ_SNDTIMEO being set on the socket.
        ret = zmq_send( args->upstream_connection
                      , burst->data
                      , burst->length
                      , 0 );
        // I belive that this is reduntant due to ZMQ_SNDTIMEO. It is more me
        // being paranoid.
        if( ret == -1 && errno == EINTR )
                goto retry_send;
        debug_log( "Upstream zmq_send() returned %d\n", ret );

        fail_if( ret == -1
               , marquise_defer_to_file( args->deferral_file, burst );
                 free_databurst( burst );
                 return;
               , "sending message: '%s'", strerror( errno ) );

        // This will timeout also due to ZMQ_RCVTIMEO
        zmq_msg_t response;
        zmq_msg_init( &response );

        retry_recv:
        ret = zmq_msg_recv( &response, args->upstream_connection, 0 );
        if( ret == -1 && errno == EINTR )
                goto retry_recv;
        debug_log( "Upstream zmq_recv() returned %d\n", ret );

        // If a recv fails, we need to reset the connection or the state
        // machine will be out of step.
        fail_if( ret == -1
               , marquise_defer_to_file( args->deferral_file, burst );
                 free_databurst( burst );
                 int linger = 0;
                 zmq_setsockopt( args->upstream_connection
                               , ZMQ_LINGER
                               , &linger
                               , sizeof( linger ) );
                 zmq_close( args->upstream_connection );
                 args->upstream_connection = connect_upstream_socket(
                           args->upstream_context
                         , args->broker );
                 return;
               , "receiving ack: '%s'", strerror( errno ) );

        // The second failure case is that upstream sends back an error as the
        // ack
        // 
        // In this case, we do not attempt to re-send, the burst is lost forever.
        fail_if( ret > 0
               , free_databurst( burst );
                 return;
               , "recieved upstream error: '%.*s'"
               , ret
               , (char *) zmq_msg_data( &response ) );

        // This may in future contain something useful, like a hash of the
        // message that we can verify.
{'fraction_non_alphanumeric': 0.09418111753371869, 'fraction_numerical': 0.0035452793834296723, 'mean_word_length': 1.5026036644165863, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 50, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1014913', 'n_tokens_mistral': 3821, 'n_tokens_neox': 3559, 'n_words': 1003}