text:
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>

#include "../../ofi.h"
#include "sofi.h"

#include "../../utils/err.h"
#include "../../utils/cont.h"
#include "../../utils/alloc.h"
#include "../../utils/fast.h"
#include "../../core/ep.h"
#include "../../core/sock.h"

/* FSM States */
#define NN_SOFI_STATE_IDLE               1001
#define NN_SOFI_STATE_CONNECTING         1002
#define NN_SOFI_STATE_ACTIVE             1003
#define NN_SOFI_STATE_CLOSING            1004
#define NN_SOFI_STATE_CLOSED             1005

/* FSM OUT State */
#define NN_SOFI_STAGEOUT_STATE_IDLE      0
#define NN_SOFI_STAGEOUT_STATE_STAGED    1

/* FSM OUT State */
#define NN_SOFI_OUT_STATE_IDLE           0
#define NN_SOFI_OUT_STATE_ACTIVE         1

/* FSM IN Flags */
#define NN_SOFI_IN_FLAG_POSTLATER        0x01
#define NN_SOFI_IN_FLAG_NNBUSY           0x02
#define NN_SOFI_IN_FLAG_NNLATER          0x04
#define NN_SOFI_IN_FLAG_FLUSH            0x08

/* FSM Handshake State */
#define NN_SOFI_HS_STATE_LOCAL           0
#define NN_SOFI_HS_STATE_FULL            1

/* Socket states */
#define NN_SOFI_SOCKET_STATE_IDLE        0
#define NN_SOFI_SOCKET_STATE_CONNECTING  1
#define NN_SOFI_SOCKET_STATE_CONNECTED   2
#define NN_SOFI_SOCKET_STATE_CLOSING     3
#define NN_SOFI_SOCKET_STATE_DRAINING    4
#define NN_SOFI_SOCKET_STATE_CLOSED      5

/* FSM Sources */
#define NN_SOFI_SRC_ENDPOINT             1101
#define NN_SOFI_SRC_KEEPALIVE_TIMER      1102
#define NN_SOFI_SRC_SHUTDOWN_TIMER       1103

/* Timeout values */
#define NN_SOFI_TIMEOUT_HANDSHAKE        1000
#define NN_SOFI_TIMEOUT_KEEPALIVE_TICK   500
#define NN_SOFI_TIMEOUT_DRAIN            OFI_DRAIN_TIMEOUT
#define NN_SOFI_TIMEOUT_SHUTDOWN         500

/* Ingress Buffer Flags */
#define NN_SOFI_INGRESS_ANCILLARY        1

/* How many ticks to wait before sending
   an keepalive packet to remote end. */
#define NN_SOFI_KEEPALIVE_OUT_TICKS      2

/* How many ticks to wait for any incoming
   message (assumed keepalive) from remote end */
#define NN_SOFI_KEEPALIVE_IN_TICKS       4

/* Memory registration keys */
#define NN_SOFI_MRM_AUX_KEY             0x0000
#define NN_SOFI_MRM_SEND_KEY            0x0001
#define NN_SOFI_MRM_RECV_KEY            0x0800
#define NN_SOFI_MRM_KEY_PAGE_SIZE       0x1000

/* Forward Declarations */
static void nn_sofi_handler (struct nn_fsm *self, int src, int type,
    void *srcptr);
static void nn_sofi_shutdown (struct nn_fsm *self, int src, int type,
    void *srcptr);
static void nn_sofi_poller_thread (void *arg);

/* Pipe definition */
static int nn_sofi_send (struct nn_pipebase *self, struct nn_msg *msg);
static int nn_sofi_recv (struct nn_pipebase *self, struct nn_msg *msg);
const struct nn_pipebase_vfptr nn_sofi_pipebase_vfptr = {
    nn_sofi_send,
    nn_sofi_recv
};

/* ########################################################################## */
/*  Utility Functions                                                         */
/* ########################################################################## */

/**
 * A critical SOFI error occured, that must result to the termination
 * of the connection.
 */
static void nn_sofi_critical_error( struct nn_sofi * self, int error )
{
    _ofi_debug("OFI[S]: Unrecoverable error #%i: %s\n", error,
        fi_strerror((int) -error));
    /* Set error & stop the FSM */
    self->error = error;
    nn_fsm_stop( &self->fsm );
}

/* ########################################################################## */
/*  Egress Functions                                                          */
/* ########################################################################## */

/**
 * Check if there are no outstanding items on the egress queue
 */
static int nn_sofi_egress_empty( struct nn_sofi * self )
{
    return self->stageout_counter.n == self->egress_max;
}

/**
 * Custom chunk deallocator function that also hints the memory registration
 * manager for the action.
 */
static void nn_sofi_mr_free( void *p, void *user )
{
    struct nn_sofi_out_ctx * ctx = user;
}

/**
 * Send the contents of the aux buffer
 */
static int nn_sofi_egress_post_aux( struct nn_sofi * self, size_t len )
{
    int ret;
    void * desc[0];
    struct iovec iov [1];
    struct fi_msg msg;

    /* Set desc */
    desc[0] = fi_mr_desc( self->aux_mr );

    /* Prepare msg IOVs */
    iov[0].iov_base = self->aux_buf;
    iov[0].iov_len = len;
    msg.desc = &desc[0];
    msg.msg_iov = iov;
    msg.iov_count = 1;
    msg.context = &self->aux_context;

    /* Send data, without generating a CQ */
    _ofi_debug("OFI[S]: Sending AUX egress (len=%zu)\n", len);
    ret = ofi_sendmsg( self->ep, &msg, 0 );
    if (ret) {
        FT_PRINTERR("ofi_sendmsg", ret);
        return ret;
    }

    /* Success */
    return 0;

}

/**
 * Get a free context that can be used to keep track of an egress packet
 */
static int nn_sofi_egress_get_context( struct nn_sofi * self,
    struct nn_sofi_out_ctx ** ctx )
{

    /* Free items should have mr_handle set to null */
    if (nn_slow( self->egress_ctx_head->mr_handle != NULL )) {
        *ctx = NULL;
        return -EAGAIN;
    }

    /* Get the head */
    _ofi_debug("OFI[S]: get_out_context: Got free ctx=%p\n", *ctx);
    *ctx = self->egress_ctx_head;
    return 0;

}

/**
 * Mark an egress context as busy
 */
static void nn_sofi_egress_mark_busy( struct nn_sofi * self,
    struct nn_sofi_out_ctx * ctx )
{
    nn_assert(nn_fast( ctx == self->egress_ctx_head ));
    nn_assert(nn_fast( ctx->mr_handle != NULL ));

    /* Add item on tail */
    _ofi_debug("OFI[S]: mark_out_busy: Moving ctx=%p to tail\n", ctx);
    ctx->prev = self->egress_ctx_tail;
    self->egress_ctx_tail->next = ctx;
    self->egress_ctx_tail = ctx;

    /* Remove item from head */
    ctx->next->prev = NULL;
    self->egress_ctx_head = ctx->next;
    ctx->next = NULL;
}

/**
 * Mark an egress context as free
 */
static void nn_sofi_egress_mark_free( struct nn_sofi * self,
    struct nn_sofi_out_ctx * ctx )
{
    nn_assert(nn_fast( ctx->mr_handle == NULL ));

    if (ctx == self->egress_ctx_tail) {

        /* Tail Item -> Move to Head */

        /* Remove item from tail */
        _ofi_debug("OFI[S]: mark_out_free: Popping ctx=%p to head\n", ctx);
        self->egress_ctx_tail = ctx->prev;
        self->egress_ctx_tail->next = NULL;
        ctx->prev = NULL;

        /* Add item on head */
        ctx->next = self->egress_ctx_head;
        self->egress_ctx_head->prev = ctx;
        self->egress_ctx_head = ctx;

    } else if (ctx == self->egress_ctx_head) {

        /* Head Item -> (Nothing) */

    } else {

        /* Mid Item -> Move to Head */

        /* Remove item from list */
        _ofi_debug("OFI[S]: mark_out_free: Moving ctx=%p to head\n", ctx);
        ctx->next->prev = ctx->prev;
        ctx->prev->next = ctx->next;

        /* Add item on head */
        ctx->next = self->egress_ctx_head;
        self->egress_ctx_head->prev = ctx;
        self->egress_ctx_head = ctx;
        ctx->prev = NULL;

    }

}

/**
 * Free an egress context
 */
static void nn_sofi_egress_free_context( struct nn_sofi * self,
    struct nn_sofi_out_ctx * ctx )
{
    _ofi_debug("OFI[S]: Freeing context=%p\n", ctx);

    /* Reset message */
    nn_msg_term(&ctx->msg);
    nn_msg_init(&ctx->msg, 0);

    /* Free memory region if we have it's handle */
    if (ctx->mr_handle) {
        _ofi_debug("OFI[S]: Freeing mr handle=%p\n", ctx->mr_handle);
        ofi_mr_release( ctx->mr_handle );
        ctx->mr_handle = NULL;
    }

}

/**
 * Post output buffers (AKA "send data"), and return
 * the number of bytes sent or the error occured.
 */
static int nn_sofi_egress_post_buffers( struct nn_sofi * self,
    struct nn_msg * outmsg )
{
    int ret;
    struct fi_msg msg;
    struct iovec iov [2];
    struct nn_sofi_out_ctx * ctx;

    /* Get a free transit context */
    ret = nn_sofi_egress_get_context( self, &ctx );
    if (ret) {
        FT_PRINTERR("nn_sofi_egress_get_context", ret);
        return ret;
    }

    /* Move message in context */
    nn_msg_mv(&ctx->msg, outmsg);

    /* Get SP Header length */
    iov [0].iov_len = nn_chunkref_size (&ctx->msg.sphdr);

    /* If SP Header is empty, use only 1 iov */
    if (nn_fast( iov[0].iov_len == 0 )) {

        /* Prepare Body IOVs */
        iov [0].iov_base = nn_chunkref_data (&ctx->msg.body);
        iov [0].iov_len = nn_chunkref_size (&ctx->msg.body);

        /* Prepare message */
        msg.msg_iov = iov;
        msg.iov_count = 1;

        /* TODO: Register a custom free function in the
                 body chunk and call ofi_mr_invalidate when
                 the user frees the chunk. */

        _ofi_debug("OFI[S]: Sending BODY[%zu]\n", iov[0].iov_len);

    } else {

        /* Prepare SP-Header + Body IOVs */
        iov [0].iov_base = nn_chunkref_data (&ctx->msg.sphdr);
        iov [1].iov_base = nn_chunkref_data (&ctx->msg.body);
        iov [1].iov_len = nn_chunkref_size (&ctx->msg.body);

        /* Prepare message */
        msg.msg_iov = iov;
        msg.iov_count = 2;

        /* TODO: Register a custom free function in the
                 body chunk and call ofi_mr_invalidate when
                 the user frees the chunk. SPHeader is
                 small enough to be copied in an MR slab */

        _ofi_debug("OFI[S]: Sending SPHDR[%zu]+BODY[%zu]\n",
            iov[0].iov_len, iov[1].iov_len);

    }

    /* Keep msg context */
    msg.context = &ctx->context;

    /* Populate MR descriptions through MRM */
    ret = ofi_mr_describe( &self->mrm_egress, &msg, &ctx->mr_handle );
    if (ret) {
        FT_PRINTERR("ofi_mr_describe", ret);
        nn_sofi_egress_free_context( self, ctx );
        return ret;
    }

    /* Send Data and generate CQ upon completed transmission */
    ret = ofi_sendmsg( self->ep, &msg, FI_COMPLETION );
    if (ret) {
        FT_PRINTERR("ofi_sendmsg", ret);
        nn_sofi_egress_free_context( self, ctx );
        return ret;
    }

    /* Mark context as busy */
    nn_sofi_egress_mark_busy( self, ctx );

    /* Return */
    return 0;
}

/**
 * Handle an error egress event (mainly memory releasing)
 */
static void nn_sofi_egress_handle_error( struct nn_sofi * self,
    struct fi_cq_err_entry * cq_entry )
{

    /* Release associated resources */
    struct nn_sofi_out_ctx * ctx = nn_cont( cq_entry->op_context,
        struct nn_sofi_out_ctx, context );
    nn_sofi_egress_free_context( self, ctx );
    nn_sofi_egress_mark_free( self, ctx );

}

/**
 * Acknowledge the fact that the ougoing data are sent
 */
static void nn_sofi_egress_handle( struct nn_sofi * self,
    struct fi_cq_msg_entry * cq_entry )
{
    int c;

    /* Reset keepalive timer */
    self->ticks_out = 0;

    /* Release associated resources */
    struct nn_sofi_out_ctx * ctx = nn_cont( cq_entry->op_context,
        struct nn_sofi_out_ctx, context );
    nn_sofi_egress_free_context( self, ctx );
    nn_sofi_egress_mark_free( self, ctx );

    /* Release back-pressure */
    c = nn_atomic_inc( &self->stageout_counter, 1);
    if (c == 0) {
        /* Operation was previously blocked because of back-pressure,
           call `nn_pipebase_sent` to allow further transmission operations. */
        nn_pipebase_sent( &self->pipebase );
    }

}

/**
 * Flush egress queue, discarding all pending messages
 */
static void nn_sofi_egress_flush( struct nn_sofi * self )
{

    /* Release egress contexts */
    struct nn_sofi_out_ctx * ctx = self->egress_ctx_head;
    while (ctx != NULL) {

        /* Free contexts that have an MR handle */
        if (ctx->mr_handle) {

            /* Increment stageout counters */
            nn_atomic_inc( &self->stageout_counter, 1 );

            /* Free context */
            nn_sofi_egress_free_context( self, ctx );

        } else {
            break;
        }
        ctx = ctx->next;
    }

}

/**
 * Send staged data
 */
static int nn_sofi_egress_send( struct nn_sofi * self )
{
    int ret;
    nn_assert(self->out_state == NN_SOFI_OUT_STATE_ACTIVE);
    nn_assert(self->stageout_state == NN_SOFI_STAGEOUT_STATE_STAGED);

    /* Post egress buffers */
    ret = nn_sofi_egress_post_buffers( self, &self->outmsg );
    if (ret) {
        FT_PRINTERR("nn_sofi_egress_post_buffers", ret);
        return ret;
    }

    /* Release staged data */
    self->stageout_state = NN_SOFI_STAGEOUT_STATE_IDLE;

    /* Apply back-pressure */
    if (nn_atomic_dec( &self->stageout_counter, 1) > 1) {

        /* Data are sent, unlock pipebase for next request */
        nn_pipebase_sent( &self->pipebase );

    } else {
        _ofi_debug("OFI[S]: Back-pressure from egress queue\n");
    }

    /* Success */
    return 0;
}

/**
 * Stage data for outgoing transmission
 */
static int nn_sofi_egress_stage( struct nn_sofi * self,
    struct nn_msg * msg )
{
    int ret;
    nn_assert( self->stageout_state == NN_SOFI_STAGEOUT_STATE_IDLE );

    /* If we are shutting down, don't accept staged messages */
    if (self->state == NN_SOFI_STATE_CLOSING)
        return 0;

    /* Move the message to the local storage. */
    nn_msg_mv (&self->outmsg, msg);
    self->stageout_state = NN_SOFI_STAGEOUT_STATE_STAGED;

    /* Check if we can send right away */
    if (nn_fast( self->out_state == NN_SOFI_OUT_STATE_ACTIVE )) {
        ret = nn_sofi_egress_send( self );
        if (ret) {
            FT_PRINTERR("nn_sofi_egress_send", ret);
            nn_sofi_critical_error( self, ret );
            return 0;
        }
    }

    /* Success */
    return 0;
}

/* ########################################################################## */
/*  Ingress Functions                                                         */
/* ########################################################################## */

/**
 * Check if there are no outstanding items on the ingress queue
 */
static int nn_sofi_ingress_empty( struct nn_sofi * self )
{
    return (self->ingress_buf_pop_head == NULL) &&
           (self->ingress_buf_busy == NULL);
}

/**
 * Return the first available populated ingress buffer (to be handled)
 */
static int nn_sofi_ingress_pop_populated( struct nn_sofi * self,
    struct nn_sofi_in_buf ** buf )
{
    /* Return EAGAIN if empty */
    if (!self->ingress_buf_pop_head) {
        _ofi_debug("OFI[S]: pop_in_populated: No items on queue\n");
        return -EAGAIN;
    }

    /* Pop populated item */
    *buf = self->ingress_buf_pop_head;
    _ofi_debug("OFI[S]: pop_in_populated: Popping buf=%p\n", *buf);
    self->ingress_buf_pop_head = self->ingress_buf_pop_head->next;
    if (self->ingress_buf_pop_head)
        self->ingress_buf_pop_head->prev = NULL;

    /* If that was the last item, also reset tail */
    if (!self->ingress_buf_pop_head) {
        _ofi_debug("OFI[S]: pop_in_populated: This was last item\n");
        self->ingress_buf_pop_tail = NULL;
    }

    /* Detach item */
    (*buf)->prev = NULL;
    (*buf)->next = NULL;

    /* Success */
    return 0;
}

/**
 * Return the first available free ingress buffer (to be posted)
 */
static int nn_sofi_ingress_pop_free( struct nn_sofi * self,
    struct nn_sofi_in_buf ** buf )
{
    /* Return EAGAIN if empty */
    if (!self->ingress_buf_free) {
        _ofi_debug("OFI[S]: pop_in_free: No items on queue\n");
        return -EAGAIN;
    }

    /* Pop free item */
    *buf = self->ingress_buf_free;
    _ofi_debug("OFI[S]: pop_in_free: Popping buf=%p\n", *buf);
    self->ingress_buf_free = self->ingress_buf_free->next;
    if (self->ingress_buf_free)
        self->ingress_buf_free->prev = NULL;

    /* Detach item */
    (*buf)->prev = NULL;
    (*buf)->next = NULL;

    /* Success */
    return 0;
}

/**
 * Mark specified buffer as free (avalable to be posted)
 */
static void nn_sofi_ingress_mark_free( struct nn_sofi * self,
    struct nn_sofi_in_buf * buf )
{

    /* If this was a 'busy' item, it was in the busy linked list,
       remove it from there. */
    if (self->ingress_buf_busy == buf) {
        /* If we were the head, pop us */
        _ofi_debug("OFI[S]: mark_in_free: Shifting buf=%p from busy\n", buf);
        self->ingress_buf_busy = self->ingress_buf_busy->next;
        if (self->ingress_buf_busy)
            self->ingress_buf_busy->prev = NULL;
        buf->prev = NULL;
        buf->next = NULL;
    } else if (buf->prev || buf->next) {
        /* If we are somewhere in the middle, remove us */
        _ofi_debug("OFI[S]: mark_in_free: Deleting buf=%p from busy\n", buf);
        if (buf->next) buf->next->prev = buf->prev;
        if (buf->prev) buf->prev->next = buf->next;
    }

    /* If we already have an item, put it first */
    if (self->ingress_buf_free) {
        /* Unshift on free queue */
        _ofi_debug("OFI[S]: mark_in_free: Unshifting buf=%p at list\n", buf);
        buf->next = self->ingress_buf_free;
        self->ingress_buf_free->prev = buf;
        self->ingress_buf_free = buf;
        buf->prev = NULL;
    } else {
        /* Make this item first buffer */
        _ofi_debug("OFI[S]: mark_in_free: Making buf=%p first item\n", buf);
        self->ingress_buf_free = buf;
        buf->next = NULL;
        buf->prev = NULL;
    }

}

/**
 * Mark specified buffer as busy (in transit)
 */
static void nn_sofi_ingress_mark_busy( struct nn_sofi * self,
    struct nn_sofi_in_buf * buf )
{
    /* If we already have an item, put it first */
    if (self->ingress_buf_busy) {

        /* Unshift on free queue */
        _ofi_debug("OFI[S]: mark_in_busy: Unshifting buf=%p at list\n", buf);
        buf->next = self->ingress_buf_busy;
        self->ingress_buf_busy->prev = buf;
        self->ingress_buf_busy = buf;
        buf->prev = NULL;

    } else {
        /* Make this item first buffer */
        _ofi_debug("OFI[S]: mark_in_busy: Making buf=%p first item\n", buf);
        self->ingress_buf_busy = buf;
        buf->next = NULL;
        buf->prev = NULL;
    }

}

/**
 * Mark specified buffer as populated (with data)
 */
static void nn_sofi_ingress_mark_populated( struct nn_sofi * self,
    struct nn_sofi_in_buf * buf )
{

    /* If this was a 'busy' item, it was in the busy linked list,
       remove it from there. */
    if (self->ingress_buf_busy == buf) {
        /* If we were the head, pop us */
        _ofi_debug("OFI[S]: mark_in_pop: Shifting buf=%p from busy\n", buf);
        self->ingress_buf_busy = self->ingress_buf_busy->next;
        if (self->ingress_buf_busy)
            self->ingress_buf_busy->prev = NULL;
        buf->prev = NULL;
        buf->next = NULL;
    } else if (buf->prev || buf->next) {
        /* If we are somewhere in the middle, remove us */
        _ofi_debug("OFI[S]: mark_in_pop: Deleting buf=%p from busy\n", buf);
        if (buf->next) buf->next->prev = buf->prev;
        if (buf->prev) buf->prev->next = buf->next;
    }

    /* If we already have an item, put it first */
    if (self->ingress_buf_pop_tail) {

        /* Put item on tail */
        _ofi_debug("OFI[S]: mark_in_pop: Unshifting buf=%p at list\n", buf);
        buf->prev = self->ingress_buf_pop_tail;
        self->ingress_buf_pop_tail->next = buf;
        self->ingress_buf_pop_tail = buf;
        buf->next = NULL;

    } else {
        /* Make this item first buffer */
        _ofi_debug("OFI[S]: mark_in_pop: Making buf=%p first item\n", buf);
        self->ingress_buf_pop_head = buf;
        self->ingress_buf_pop_tail = buf;
        buf->next = NULL;
        buf->prev = NULL;
    }

}

/**
 * Flush ingress queue, discarding all pending messages
 */
static void nn_sofi_ingress_flush( struct nn_sofi * self )
{
    struct nn_sofi_in_buf * buf;

    /* Mark all items on busy queue as 'free', effectively ignoring
       all messages pending in the rx queue. */
    while (self->ingress_buf_busy) {
        nn_sofi_ingress_mark_free( self, self->ingress_buf_busy );
    }

    /* If there is a pending nanomsg operation, don't flush now */
    if (self->ingress_flags & NN_SOFI_IN_FLAG_NNBUSY) {

        /* Mark ingress queue for delayed flushing */
        self->ingress_flags |= NN_SOFI_IN_FLAG_FLUSH;
        _ofi_debug("OFI[S]: Nanomsg is busy, delaying ingress flush\n");
        return;

    }

    /* Discard all receiving messages */
    while (self->ingress_buf_pop_head) {
        nn_sofi_ingress_pop_populated( self, &buf );
        nn_sofi_ingress_mark_free( self, buf );
    }

    /* Discard all items on ingress buffer */
    self->ingress_flags &= ~NN_SOFI_IN_FLAG_POSTLATER;
    self->ingress_flags &= ~NN_SOFI_IN_FLAG_NNLATER;

}

/**
 * Post a sofi buffer as ingress
 */
static int nn_sofi_ingress_post_buffer( struct nn_sofi * self,
    struct nn_sofi_in_buf * buf )
{
    int ret;
    struct iovec iov[1];
    struct fi_msg msg;

    /* Keep index of active buffer */
    _ofi_debug("OFI[S]: Posting ingress buffer=%p\n", nn_chunk_deref( buf->chunk ));

    /* Prepare message from active ingress buffer */
    memset( &msg, 0, sizeof(msg) );
    iov[0].iov_base = nn_chunk_deref( buf->chunk );
    iov[0].iov_len = self->ingress_buf_size;
    msg.desc = &buf->mr_desc[0];
    msg.msg_iov = &iov[0];
    msg.iov_count = 1;
    msg.context = &buf->context;

    /* Post receive buffers */
    ret = ofi_recvmsg( self->ep, &msg, 0 );
    if (ret) {

        /* Mark buffer as free */
        nn_sofi_ingress_mark_free( self, buf );

        /* Return error */
        FT_PRINTERR("ofi_recvmsg", ret);
        return ret;

    }

    /* Mark buffer as busy */
    nn_sofi_ingress_mark_busy( self, buf );

    /* Success */
    return 0;

}

/**
 * Post all ingress messages
 */
static int nn_sofi_ingress_post_all( struct nn_sofi * self )
{
    int ret;
    struct nn_sofi_in_buf * buf;

    /* Post ingress buffers until we ran ount of items */
    while (self->ingress_buf_free) {

        /* Pop free item */
        ret = nn_sofi_ingress_pop_free( self, &buf );
        if (ret) {
            FT_PRINTERR("nn_sofi_ingress_pop_free", ret);
            _ofi_debug("OFI[S]: Failed to get next free buffer!\n");
            return ret;
        }

        /* Post free item */
        ret = nn_sofi_ingress_post_buffer( self, buf );
        if (ret) {
            FT_PRINTERR("nn_sofi_ingress_post_buffer", ret);
            _ofi_debug("OFI[S]: Failed to post all ingress buffers!\n");
            return ret;
        }
    }

    /* Success */
    return ret;
}

/**
 * Post receive buffers
 *
 * This function should pick one of the available pre-allocated message buffers
 * and post them to libfabric. After that, we are expecting a CQ event to
 * trigger the `sofi_ingress_handle` in order to receive the incoming data.
 */
static int nn_sofi_ingress_post( struct nn_sofi * self )
{
    int ret;
    struct nn_sofi_in_buf * buf;

    /* Get the input buffer to post */
    ret = nn_sofi_ingress_pop_free( self, &buf );
    if (ret == -EAGAIN) {
        _ofi_debug("OFI[S]: No free ingress buffers found\n");
        return -EAGAIN;
    } else if (ret) {
        FT_PRINTERR("nn_sofi_ingress_pop_free", ret);
        return ret;
    }

    /* Post the input buffer */
    ret = nn_sofi_ingress_post_buffer( self, buf );
    if (ret < 0) {
        FT_PRINTERR("nn_sofi_ingress_post_buffer", ret);
        return ret;
    }

    /* Success */
    return 0;
}

/**
 * Handle an ingress error event
 */
static void nn_sofi_ingress_handle_error( struct nn_sofi * self,
    struct fi_cq_err_entry * cq_entry )
{
    /* Nothing really to do here */
}

/**
 * Process input data
 *
 * Upon completion, this function should return 0 if there are input free
 * buffers available for re-posting or -EAGAIN otherwise. In case an error
 * occurs, this function will return the appropriate error code.
 */
static void nn_sofi_ingress_handle( struct nn_sofi * self,
    struct fi_cq_msg_entry * cq_entry )
{
    struct nn_sofi_in_buf * buf;

    /* Reset keepalive timer */
    self->ticks_in = 0;

    /* Get the posted ingress buffer */
    buf = nn_cont( cq_entry->op_context, struct nn_sofi_in_buf, context );
    _ofi_debug("OFI[S]: Handling ingress=%i\n",
        (int)(buf - self->ingress_buffers));

    /* Check if this is a keepalive message */
    if (cq_entry->len == NN_SOFI_KEEPALIVE_PACKET_LEN) {

        /* Test payload contents */
        if (memcmp( nn_chunk_deref(buf->chunk), NN_SOFI_KEEPALIVE_PACKET,
            NN_SOFI_KEEPALIVE_PACKET_LEN ) == 0)
        {

            /* Mark buffer as free */
            _ofi_debug("OFI[S]: Received KEEPALIVE\n");
            nn_sofi_ingress_mark_free( self, buf );

            /* If we have a POSTLATER flag, try again now */
            if (self->ingress_flags & NN_SOFI_IN_FLAG_POSTLATER) {

                /* Post another ingress buffer */
                _ofi_debug("OFI[S]: Posting late ingress buffers\n");
                self->ingress_flags &= ~NN_SOFI_IN_FLAG_POSTLATER;
                nn_sofi_ingress_post( self );

            }

            /* No need to continue */
            return;

        }

    }

    /* Prepare message from the active staged buffer */
    _ofi_debug("OFI[S]: Received BODY[%zu]\n", cq_entry->len);
    nn_chunk_reset( buf->chunk, cq_entry->len );
    nn_msg_init_chunk( &buf->msg, buf->chunk );
    nn_chunk_addref( buf->chunk, 1 );

    /* Mark buffer as populated */
    nn_sofi_ingress_mark_populated( self, buf );

    /* If nanomsg is busy, try later */
    if (self->ingress_flags & NN_SOFI_IN_FLAG_NNBUSY) {
        _ofi_debug("OFI[S]: NanoMsg is busy, will try later\n");
        self->ingress_flags |= NN_SOFI_IN_FLAG_NNLATER;
    } else {
        _ofi_debug("OFI[S]: Notifying NanoMsg\n");
        self->ingress_flags |= NN_SOFI_IN_FLAG_NNBUSY;
        nn_pipebase_received( &self->pipebase );
    }

}

/**
 * Pop a message from the ingress queue
 */
static int nn_sofi_ingress_fetch( struct nn_sofi * self,
    struct nn_msg * msg )
{
    int ret;
    struct nn_sofi_in_buf * buf;
    nn_assert(nn_fast( self->ingress_flags & NN_SOFI_IN_FLAG_NNBUSY ));

    /* Pick first busy buffer to receive */
    ret = nn_sofi_ingress_pop_populated( self, &buf );
    if (ret) {
        FT_PRINTERR("nn_sofi_ingress_pop_populated", ret);
        _ofi_debug("OFI[S]: Unable to fetch next populated buffer\n");
        return ret;
    }

    /* Move message to output */
    _ofi_debug("OFI[S]: Passing to nanomsg ingress buffer=%i\n", ret);
    nn_msg_mv( msg, &buf->msg );

    /* Mark buffer as free */
    nn_sofi_ingress_mark_free( self, buf );

    /* If we have a POSTLATER flag, post now */
    if (self->ingress_flags & NN_SOFI_IN_FLAG_POSTLATER) {

        /* Post another ingress buffer */
        _ofi_debug("OFI[S]: Posting late ingress buffers\n");
        self->ingress_flags &= ~NN_SOFI_IN_FLAG_POSTLATER;
        nn_sofi_ingress_post( self );

    }

    /* We are not busy any more */
    self->ingress_flags &= ~NN_SOFI_IN_FLAG_NNBUSY;

    /* Check if we should post another nanomsg trigger */
    if (self->ingress_flags & NN_SOFI_IN_FLAG_NNLATER) {
        _ofi_debug("OFI[S]: Notifying NanoMsg (later)\n");

        /* Reset flags */
        self->ingress_flags = (self->ingress_flags & ~NN_SOFI_IN_FLAG_NNLATER)
                            | NN_SOFI_IN_FLAG_NNBUSY;

        /* We have data */
        nn_pipebase_received( &self->pipebase );

    }

    /* If we must be flushed, retry now */
    if (self->ingress_flags & NN_SOFI_IN_FLAG_FLUSH) {
        _ofi_debug("OFI[S]: Executing delayed ingress queue flush\n");
        nn_sofi_ingress_flush( self );
    }

    /* Return success */
    return 0;
}

/* ########################################################################## */
/*  Implementation  Functions                                                 */
/* ########################################################################## */

/*  Initialize the state machine */
int nn_sofi_init ( struct nn_sofi *self, struct ofi_domain *domain, int offset,
    struct nn_epbase *epbase, int src, struct nn_fsm *owner )
{
    const uint64_t mr_flags = FI_RECV | FI_READ | FI_REMOTE_WRITE;
    uint64_t mr_page_offset = NN_SOFI_MRM_KEY_PAGE_SIZE * offset;
    int ret, i;

    /* Initialize properties */
    self->domain = domain;
    self->ep = NULL;
    self->offset = offset;
    self->epbase = epbase;
    self->stageout_state = NN_SOFI_STAGEOUT_STATE_IDLE;
    self->out_state = NN_SOFI_OUT_STATE_IDLE;

    /* Initialize local handshake */
    self->hs_local.version = 1;

#ifndef OFI_DISABLE_HANDSHAKE
    self->hs_state = NN_SOFI_HS_STATE_LOCAL;
#endif

    /* ----------------------------------- */
    /*  NanoMSG Core Initialization        */
    /* ----------------------------------- */

    /* Initialize list item */
    nn_list_item_init (&self->item);

    /* Initialize pipe base */
    nn_pipebase_init (&self->pipebase, &nn_sofi_pipebase_vfptr, epbase);

    /* Initialize FSM */
    nn_fsm_init (&self->fsm, nn_sofi_handler, nn_sofi_shutdown,
        src, self, owner);

    /* Reset properties */
    self->state = NN_SOFI_STATE_IDLE;
    self->socket_state = NN_SOFI_STATE_IDLE;
    self->error = 0;

    /* ----------------------------------- */
    /*  NanoMsg Component Initialization   */
    /* ----------------------------------- */

    /* Initialize timers */
    nn_timer_init(&self->timer_keepalive, NN_SOFI_SRC_KEEPALIVE_TIMER,
        &self->fsm);
    nn_timer_init(&self->timer_shutdown, NN_SOFI_SRC_SHUTDOWN_TIMER,
        &self->fsm);

    /* Reset properties */
    self->ticks_in = 0;
    self->ticks_out = 0;

    /* Outgoing message */
    nn_msg_init (&self->outmsg, 0);

    /* ----------------------------------- */
    /*  OFI Sub-Component Initialization   */
    /* ----------------------------------- */

    /* Get options */
    int rx_queue, tx_queue, rx_msg_size, slab_size;
    size_t opt_sz = sizeof(int);
    nn_epbase_getopt (epbase, NN_OFI, NN_OFI_TX_QUEUE_SIZE, &tx_queue, &opt_sz);
    nn_epbase_getopt (epbase, NN_OFI, NN_OFI_RX_QUEUE_SIZE, &rx_queue, &opt_sz);
    nn_epbase_getopt (epbase, NN_OFI, NN_OFI_SLAB_SIZE, &slab_size, &opt_sz);
    nn_epbase_getopt (epbase, NN_SOL_SOCKET, NN_RCVBUF, &rx_msg_size, &opt_sz);

    /* Put default values if set to AUTO */
    if (tx_queue == 0) tx_queue = domain->fi->tx_attr->size;
    if (rx_queue == 0) rx_queue = domain->fi->rx_attr->size;

    /* Wrap overflown values */
    if (tx_queue > domain->fi->tx_attr->size) tx_queue = domain->fi->tx_attr->size;
    if (rx_queue > domain->fi->rx_attr->size) rx_queue = domain->fi->rx_attr->size;

    /* Debug print current values */
    _ofi_debug("OFI[S]: Options: Tx-Queue-Size: %i"
                    ", Rx-Queue-Size: %i, Offset: %i\n",
                    rx_queue, tx_queue, offset);
    _ofi_debug("OFI[S]:          Max-Recv-Size: %i b, Slab-Size: %i b\n",
                    rx_msg_size, slab_size);

    /* ####[ ANCILLARY ]#### */

    /* Register ancillary data */
    ret = fi_mr_reg(self->domain->domain, self->aux_buf, NN_SOFI_ANCILLARY_SIZE,
        FI_RECV| FI_READ| FI_REMOTE_WRITE| FI_SEND| FI_WRITE| FI_REMOTE_READ,
        0, mr_page_offset+NN_SOFI_MRM_AUX_KEY, 0, &self->aux_mr, NULL);
    if (ret) {
        FT_PRINTERR("fi_mr_reg", ret);
    }

    /* Currently AUX buffer is only used for keepalive message,
       so for optimisation reasons, write it's contents once now */
    memcpy( self->aux_buf, NN_SOFI_KEEPALIVE_PACKET,
        NN_SOFI_KEEPALIVE_PACKET_LEN );

    /* ####[ EGRESS ]#### */

    /* Get an OFI worker */
    self->worker = ofi_fabric_getworker( domain->parent, &self->fsm );

    /* Initialize egress MR Manager with 32 banks */
    struct ofi_mr_bank_attr mrattr_tx = {
         /* Worst case Tx scenario : Body bank + SP Header bank */
        .bank_count = tx_queue * 2,
        .domain = self->domain,
        .direction = OFI_MR_DIR_SEND,
        .slab_count = tx_queue,
        .slab_size = slab_size,
        .base_key = mr_page_offset+NN_SOFI_MRM_SEND_KEY,
    };
    ofi_mr_init( &self->mrm_egress, &mrattr_tx );

    /* Initialize throttle counters */
    nn_atomic_init( &self->stageout_counter, tx_queue );
    self->egress_max = tx_queue;

    /* A list of egress chunks */
    self->egress_contexts = nn_alloc( sizeof(struct nn_sofi_out_ctx ) * tx_queue,
        "egress sofi buffers");
    nn_assert( self->egress_contexts );
    memset( self->egress_contexts, 0, sizeof(struct nn_sofi_out_ctx) * tx_queue);

    /* Initilaize buffers */
    for (i=0; i<rx_queue; ++i) {

        /* Initialize properties */
        self->egress_contexts[i].sofi = self;
        nn_msg_init( &self->egress_contexts[i].msg, 0);

        /* Setup linked list */
        if (i > 0) {
            self->egress_contexts[i-1].next = &self->egress_contexts[i];
        }
        if (i < rx_queue-1) {
            self->egress_contexts[i+1].prev = &self->egress_contexts[i];
        }

    }

    /* Setup ring */
    self->egress_ctx_head = &self->egress_contexts[0];
    self->egress_ctx_tail = &self->egress_contexts[rx_queue-1];

    /* ####[ INGRESS ]#### */

    /* Allocate ingress buffers */
    self->ingress_buffers = nn_alloc( sizeof(struct nn_sofi_in_buf) * rx_queue,
        "ingress sofi buffer" );
    nn_assert( self->ingress_buffers );
    memset( self->ingress_buffers, 0, sizeof(struct nn_sofi_in_buf) * rx_queue);

    /* Initilaize buffers */
    for (i=0; i<rx_queue; ++i) {

        /* Allocate chunk */
        ret = nn_chunk_alloc( rx_msg_size, NN_ALLOC_PAGEALIGN,
            &self->ingress_buffers[i].chunk );
        if (ret == -ENOSYS) {
            /* Page-aligned allocator failed, use default */
            ret = nn_chunk_alloc( rx_msg_size, 0,
                &self->ingress_buffers[i].chunk );
            _ofi_debug("OFI[S]: Allocated non-aligned ingress chunk=%p\n",
                self->ingress_buffers[i].chunk);
        } else {
            _ofi_debug("OFI[S]: Allocated aligned ingress chunk=%p\n",
                self->ingress_buffers[i].chunk);
        }
        if (ret) {
            FT_PRINTERR("nn_chunk_alloc", ret);
            return ret;
        }

        /* Register memory */
        ret = fi_mr_reg(self->domain->domain, self->ingress_buffers[i].chunk,
            rx_msg_size, mr_flags, 0, mr_page_offset+NN_SOFI_MRM_RECV_KEY+i, 0,
            &self->ingress_buffers[i].mr, NULL);
        if (ret) {
            FT_PRINTERR("fi_mr_reg", ret);
        }

        /* Get desc */
        self->ingress_buffers[i].mr_desc[0] =
            fi_mr_desc( self->ingress_buffers[i].mr );

        /* Implement bi-directional linked list */
        if (i > 0) {
            self->ingress_buffers[i-1].next = &self->ingress_buffers[i];
        }
        if (i < rx_queue-1) {
            self->ingress_buffers[i+1].prev = &self->ingress_buffers[i];
        }

    }

    /* Initial linked list values */
    self->ingress_buf_free = &self->ingress_buffers[0];
    self->ingress_buf_busy = NULL;
    self->ingress_buf_pop_head = NULL;
    self->ingress_buf_pop_tail = NULL;
    self->ingress_flags = 0;

    /* Populate ingress properties */
    self->ingress_buf_size = rx_msg_size;
    self->ingress_max = rx_queue;

    /* Success */
    return 0;

}

/**
 * Start SOFI on the accepting side
 */
int nn_sofi_start_accept( struct nn_sofi *self, struct fi_eq_cm_entry * conreq )
{
    int ret;

    /* Disable worker */
    nn_ofiw_stop( self->worker );

    /* Open active endpoint */
    ret = ofi_active_endpoint_open( self->domain, self->worker,
        NN_SOFI_SRC_ENDPOINT, NULL, conreq->info, &self->ep );
    if (ret) {
        FT_PRINTERR("ofi_active_endpoint_open", ret);
        return ret;
    }

    /* Receive remote handshake information */
#ifndef OFI_DISABLE_HANDSHAKE
    memcpy( &self->hs_remote, conreq->data, sizeof(self->hs_remote) );
    _ofi_debug("OFI[S]: Handshake with remote version=%i\n",
        self->hs_remote.version);
    self->hs_state = NN_SOFI_HS_STATE_FULL;
#endif

    /* Accept incoming connection and send our side of the handshake */
    self->socket_state = NN_SOFI_STATE_CONNECTING;
#ifdef OFI_DISABLE_HANDSHAKE
    ret = ofi_cm_accept( self->ep, NULL, 0 );
#else
    ret = ofi_cm_accept( self->ep, &self->hs_local, sizeof(self->hs_local) );
#endif
    if (ret) {
        FT_PRINTERR("ofi_cm_accept", ret);
        self->socket_state = NN_SOFI_SOCKET_STATE_CLOSED;
        return ret;
    }

    /* Start FSM */
    _ofi_debug("OFI[S]: Starting Accepted FSM \n");
    nn_fsm_start (&self->fsm);

    /* Start worker */
    nn_ofiw_start( self->worker );

    /* Success */
    return 0;
}

/**
 * Start SOFI on the connecting side
 */
int nn_sofi_start_connect( struct nn_sofi *self )
{
    int ret;

    /* Disable worker */
    nn_ofiw_stop( self->worker );

    /* Open active endpoint */
    ret = ofi_active_endpoint_open( self->domain, self->worker,
        NN_SOFI_SRC_ENDPOINT, NULL, NULL, &self->ep );
    if (ret) {
        FT_PRINTERR("ofi_active_endpoint_open", ret);
        return ret;
    }

    /* Connect to the remote endpoint */
    self->socket_state = NN_SOFI_STATE_CONNECTING;
#ifdef OFI_DISABLE_HANDSHAKE
    ret = ofi_cm_connect( self->ep, NULL, NULL, 0 );
#else
    ret = ofi_cm_connect( self->ep, NULL, &self->hs_local,
        sizeof(self->hs_local) );
#endif
    if (ret) {
        FT_PRINTERR("ofi_cm_connect", ret);
        self->socket_state = NN_SOFI_SOCKET_STATE_CLOSED;
        return ret;
    }

    /* Start FSM */
    _ofi_debug("OFI[S]: Starting Connected FSM \n");
    nn_fsm_start (&self->fsm);

    /* Enable worker */
    nn_ofiw_start( self->worker );

    /* Success */
    return 0;
}

/**
 * Cleanup the state machine
 */
void nn_sofi_term (struct nn_sofi *self)
{
    struct nn_sofi_out_ctx *ctx;
    struct nn_list_item *it;
    int i, ret;
    _ofi_debug("OFI[S]: Cleaning-up SOFI\n");

    /* ----------------------------------- */
    /*  OFI Sub-Component Termination      */
    /* ----------------------------------- */

    /* Terminate worker */
    nn_ofiw_term( self->worker );

    /* Terminate MR manager */
    ofi_mr_term( &self->mrm_egress );

    /* Free chunks */
    for (i=0; i<self->ingress_max; ++i) {

        /* Unregister memory */
        ret = fi_close(&self->ingress_buffers[i].mr->fid);
        if (ret) {
            FT_PRINTERR("fi_mr_reg", ret);
        }

        /* Free chunk */
        nn_chunk_free( self->ingress_buffers[i].chunk );

    }

    /* Free ingress messages */
    nn_free( self->ingress_buffers );

    /* Unregister ancillary MR */
    ret = fi_close(&self->aux_mr->fid);
    if (ret) {
        FT_PRINTERR("fi_mr_reg", ret);
    }

    /* ----------------------------------- */
    /*  NanoMsg Component Termination      */
    /* ----------------------------------- */

    /* Stop timers */
    nn_timer_term (&self->timer_keepalive);
    nn_timer_term (&self->timer_shutdown);

    /* Stop throttles */
    nn_atomic_term( &self->stageout_counter );

    /* Release all transit contexts in book-keeping */
    ctx = self->egress_ctx_head;
    while (ctx != NULL) {
        nn_msg_term( &ctx->msg);
        ctx = ctx->next;
    }
    nn_free( self->egress_contexts );

    /* ----------------------------------- */
    /*  NanoMSG Core Termination           */
    /* ----------------------------------- */

    /* Terminate list item */
    nn_list_item_term (&self->item);

    /* Cleanup components */
    nn_pipebase_term (&self->pipebase);
    nn_fsm_term (&self->fsm);

}

/**
 * Check if FSM is idle
 */
int nn_sofi_isidle (struct nn_sofi *self)
{
    return nn_fsm_isidle (&self->fsm);
}

/**
 * Stop the state machine
 */
void nn_sofi_stop (struct nn_sofi *self)
{
    /* Stop FSM & Switch to shutdown handler */
    _ofi_debug("OFI[S]: Stopping FSM\n");
    nn_fsm_stop (&self->fsm);
}

/* ============================== */
/*    INTERFACE IMPLEMENTATION    */
/* ============================== */

/**
 * This function is called by the nanomsg core when some data needs to be sent.
 * It's important to call the 'nn_pipebase_sent' function when ready!
 */
static int nn_sofi_send (struct nn_pipebase *pb, struct nn_msg *msg)
{
    _ofi_debug("OFI[S]: NanoMsg SEND event\n");
    int ret;
    struct nn_sofi *self;
    self = nn_cont (pb, struct nn_sofi, pipebase);

    /* We sent something */
    self->ticks_out = 0;

    /* Push a message to the egress queue */
    return nn_sofi_egress_stage( self, msg );

}

/**
 * This function is called by the nanomsg core when some data needs to be sent.
 * This is triggered only when 'nn_pipebase_received' is called!
 */
static int nn_sofi_recv (struct nn_pipebase *pb, struct nn_msg *msg)
{
    int ret;
    struct nn_sofi *self;
    self = nn_cont (pb, struct nn_sofi, pipebase);

    /* Fetch a message from the ingress queue */
    _ofi_debug("OFI[S]: NanoMsg RECV event\n");
    return nn_sofi_ingress_fetch( self, msg );
}

/**
 * SHUTDOWN State Handler
 */
static void nn_sofi_shutdown (struct nn_fsm *fsm, int src, int type,
    void *srcptr)
{
    struct nn_sofi *self;
    struct fi_cq_msg_entry * cq_entry;
    struct fi_cq_err_entry * cq_error;

    /* Get pointer to sofi structure */
    self = nn_cont (fsm, struct nn_sofi, fsm);

    /* If this is part of the FSM action, start shutdown */
    if (nn_slow (src == NN_FSM_ACTION && type == NN_FSM_STOP)) {
        _ofi_debug("OFI[S]: We are now closing\n");

        /* Enter closing state */
        self->state = NN_SOFI_STATE_CLOSING;

        /* Stop keepalive timer */
        nn_timer_stop( &self->timer_keepalive );

        /* Flush ingress queue */
        _ofi_debug("OFI[S]: Discarding pending ingress buffers\n");
        nn_sofi_ingress_flush( self );

        /* If we are not connected, or if we are closing due to an
           error, discard all staged egress messages */
        if ((self->state == NN_SOFI_STATE_CONNECTING) || (self->error != 0)) {
            _ofi_debug("OFI[S]: Discarding pending egress buffers\n");
            nn_sofi_egress_flush( self );
        }

    } else if (nn_slow(src == (NN_SOFI_SRC_ENDPOINT | OFI_SRC_CQ_TX) )) {

        /* Handle Tx Events. Other Tx events won't be accepted because our
           state is now in NN_SOFI_STATE_CLOSING */

        /* Wait for successful egress completion only on clean shutdown */
        if (self->error == 0) {
            _ofi_debug("OFI[S]: Draining egress CQ event\n");
            if (type == NN_OFIW_COMPLETED) {
                /* Handle completed event */
                cq_entry = (struct fi_cq_msg_entry *) srcptr;
                nn_sofi_egress_handle( self, cq_entry );
            } else if (type == NN_OFIW_ERROR) {
                /* Handle error event */
                cq_error = (struct fi_cq_err_entry *) srcptr;
                nn_sofi_egress_handle_error( self, cq_error );
            } else {
                nn_fsm_bad_action (self->state, src, type);
            }
        } else {
            _ofi_debug("OFI[S]: Discarding egress CQ event due to error\n");
            if ((type != NN_OFIW_COMPLETED) && (type != NN_OFIW_ERROR)) {
                nn_fsm_bad_action (self->state, src, type);
            }
        }

    } else if (nn_slow(src == (NN_SOFI_SRC_ENDPOINT | OFI_SRC_CQ_RX) )) {

        /* Handle Rx Events, but don't post new input buffers */
        _ofi_debug("OFI[S]: Discarding ingress CQ event\n");
        if ((type != NN_OFIW_COMPLETED) && (type != NN_OFIW_ERROR)) {
            nn_fsm_bad_action (self->state, src, type);
        }

    } else if (nn_slow(src == (NN_SOFI_SRC_ENDPOINT | OFI_SRC_EQ) )) {

        /* Handle endpoint events */
        _ofi_debug("OFI[S]: Draining endpoint EQ event\n");
        if (type == FI_SHUTDOWN) {

            /* We can now safely close the socket */
            _ofi_debug("OFI[S]: Socket is closed\n");
            self->socket_state = NN_SOFI_SOCKET_STATE_CLOSED;
            nn_timer_stop( &self->timer_shutdown );

        } else {
            /* We also accept error */
            if (type > 0) nn_fsm_bad_action (self->state, src, type);
        }

    } else if (nn_slow(src == NN_SOFI_SRC_KEEPALIVE_TIMER)) {

        /* Wait for timer to stop */
        if (nn_slow( type != NN_TIMER_STOPPED ))
            nn_fsm_bad_action (self->state, src, type);

        _ofi_debug("OFI[S]: Keepalive timer stopped\n");

    } else if (nn_slow(src == NN_SOFI_SRC_SHUTDOWN_TIMER)) {

        /* Wait for timer to stop */
        if (nn_fast( type == NN_TIMER_TIMEOUT )) {
            if (self->socket_state == NN_SOFI_SOCKET_STATE_CLOSING) {
                /* Stop timer (handled at stop event) */
                nn_timer_stop( &self->timer_shutdown );
#if OFI_DRAIN_TIMEOUT > 0

            } else if (self->socket_state == NN_SOFI_SOCKET_STATE_DRAINING) {

                /* Stop timer (it will be restarted at stop event) */
                nn_timer_stop( &self->timer_shutdown );
#endif
            }
            return;

        } else if (nn_fast( type == NN_TIMER_STOPPED )) {

#if OFI_DRAIN_TIMEOUT > 0
            if (nn_fast( self->socket_state == NN_SOFI_SOCKET_STATE_DRAINING)) {
                /* Shutdown connection & close endpoint */
                _ofi_debug("OFI[S]: Stopping endpoint\n");
                self->socket_state = NN_SOFI_SOCKET_STATE_CLOSING;
                ofi_cm_shutdown( self->ep );

                /* Restart timer to track shutdown timeouts */
                nn_timer_start( &self->timer_shutdown,
                    NN_SOFI_TIMEOUT_SHUTDOWN );
                return;

            } else
#endif
            if (nn_slow( self->socket_state == NN_SOFI_SOCKET_STATE_CLOSING)) {
                /* Timed out waiting for shutdown */
                _ofi_debug("OFI[S]: Timed out waiting for shutdown EQ event\n");
                self->socket_state = NN_SOFI_SOCKET_STATE_CLOSED;

            }else if(nn_slow(self->socket_state!=NN_SOFI_SOCKET_STATE_CLOSED)) {
                nn_fsm_bad_state (self->state, src, type);
            }

        } else {
            nn_fsm_bad_action (self->state, src, type);
        }

    } else {
        nn_fsm_bad_source (self->state, src, type);
    }

    /* Wait for all outstanding transmissions or receptions to complete
       and for all resources to be stopped */
    if (!nn_sofi_egress_empty( self ) ||
        !nn_sofi_ingress_empty( self ) ||
        !nn_timer_isidle( &self->timer_keepalive ) ||
        !nn_timer_isidle( &self->timer_shutdown) ) {
        _ofi_debug("OFI[S]: Delaying close: egress=%i, ingress=%i, isidle=%i"
            ",%i, egress_left=%i\n",
            nn_sofi_egress_empty( self ), nn_sofi_ingress_empty( self ),
            nn_timer_isidle( &self->timer_keepalive ),
            nn_timer_isidle( &self->timer_shutdown ),
            self->egress_max - self->stageout_counter.n);
        return;
    }

    if (self->socket_state == NN_SOFI_SOCKET_STATE_CONNECTED) {

#if OFI_DRAIN_TIMEOUT > 0
        /* Start draining socket */
        _ofi_debug("OFI[S]: Draining pending endpoint events (%i us)\n",
            NN_SOFI_TIMEOUT_DRAIN);
        self->socket_state = NN_SOFI_SOCKET_STATE_DRAINING;
        nn_timer_start( &self->timer_shutdown,
            NN_SOFI_TIMEOUT_DRAIN );
#else
        /* Shutdown connection & close endpoint */
        _ofi_debug("OFI[S]: Stopping endpoint\n");
        self->socket_state = NN_SOFI_SOCKET_STATE_CLOSING;
        ofi_cm_shutdown( self->ep );

        /* Restart timer to track shutdown timeouts */
        nn_timer_start( &self->timer_shutdown,
            NN_SOFI_TIMEOUT_SHUTDOWN );
#endif
        return;

    } else if (self->socket_state == NN_SOFI_STATE_CONNECTING) {

        /* The socket never managed to connect */
        _ofi_debug("OFI[S]: Socket never connected\n");
        self->socket_state = NN_SOFI_SOCKET_STATE_CLOSED;

    } else if (self->socket_state != NN_SOFI_SOCKET_STATE_CLOSED) {

        _ofi_debug("OFI[S]: Unexpected socket_state=%i\n",
            self->socket_state);
        nn_fsm_bad_source (self->state, src, type);

    }

    /* Disable worker */
    nn_ofiw_stop( self->worker );

    /* Close endpoint */
    _ofi_debug("OFI[S]: Closing endpoint\n");
    ofi_active_endpoint_close( self->ep );

    /* Stop nanomsg components */
    _ofi_debug("OFI[S]: Stopping pipebase\n");
    nn_pipebase_stop (&self->pipebase);
    nn_fsm_stopped(&self->fsm, NN_SOFI_STOPPED);

}

/**
 * ACTIVE State Handler
 */
static void nn_sofi_handler (struct nn_fsm *fsm, int src, int type,
    void *srcptr)
{
    int ret;
    struct nn_sofi *self;
    struct fi_eq_cm_entry * cq_cm_entry;
    struct fi_cq_msg_entry * cq_entry;
    struct fi_cq_err_entry * cq_error;

    /* Get pointer to sofi structure */
    self = nn_cont (fsm, struct nn_sofi, fsm);
    _ofi_debug("OFI[S]: nn_sofi_handler state=%i, src=%i, type=%i\n",
        self->state, src, type);

    /* Handle state transitions */
    switch (self->state) {

/******************************************************************************/
/*  IDLE state.                                                               */
/******************************************************************************/
    case NN_SOFI_STATE_IDLE:
        switch (src) {

        /* ========================= */
        /*  FSM Action               */
        /* ========================= */
        case NN_FSM_ACTION:
            switch (type) {
            case NN_FSM_START:

                /* Wait for connection to be established before starting pipe */
                _ofi_debug("OFI[S]: FSM started, waiting for connection\n");
                self->state = NN_SOFI_STATE_CONNECTING;
                return;

            default:
                nn_fsm_bad_action (self->state, src, type);
            }

        default:
            nn_fsm_bad_source (self->state, src, type);
        }

/******************************************************************************/
/*  NN_SOFI_STATE_CONNECTING state.                                           */
/*  We are waiting a connection event form the endpoint.                      */
/******************************************************************************/
    case NN_SOFI_STATE_CONNECTING:
        switch (src) {

        /* ========================= */
        /*  Endpoint EQ Action       */
        /* ========================= */
        case NN_SOFI_SRC_ENDPOINT | OFI_SRC_EQ:
            switch (type) {
            case FI_CONNECTED:

                /* Receive remote handshake information */
                cq_cm_entry = (struct fi_eq_cm_entry *) srcptr;
#ifndef OFI_DISABLE_HANDSHAKE
                if ( self->hs_state == NN_SOFI_HS_STATE_LOCAL ) {
                    memcpy( &self->hs_remote, cq_cm_entry->data,
                        sizeof(self->hs_remote) );
                    _ofi_debug("OFI[S]: Handshake with remote version=%i\n",
                        self->hs_remote.version);
                }
#endif

                /* The connection is established, start pipe */
                _ofi_debug("OFI[S]: Endpoint connected, starting pipebase\n");
                self->state = NN_SOFI_STATE_ACTIVE;
                self->socket_state = NN_SOFI_SOCKET_STATE_CONNECTED;
                self->out_state = NN_SOFI_OUT_STATE_ACTIVE;
                nn_pipebase_start( &self->pipebase );

                /* Start keepalive timer */
                _ofi_debug("OFI[S]: Starting keepalive timer\n");
                nn_timer_start( &self->timer_keepalive,
                    NN_SOFI_TIMEOUT_KEEPALIVE_TICK );

                /* Post all input buffers */
                ret = nn_sofi_ingress_post( self );
                if (ret) {
                    FT_PRINTERR("nn_sofi_ingress_post_all", ret);
                    nn_sofi_critical_error( self, ret );
                    return;
                }

                /* Now it's time to send staged data */
                if (self->stageout_state == NN_SOFI_STAGEOUT_STATE_STAGED) {
                    ret = nn_sofi_egress_send( self );
                    if (ret) {
                        FT_PRINTERR("nn_sofi_egress_send", ret);
                        nn_sofi_critical_error( self, ret );
                    }
                }
                return;

            default:

                /* EQ Error */
                if (type < 0) {

                    /* Unrecoverable comm error */
                    nn_sofi_critical_error( self, type );
                    return;

                }

                nn_fsm_bad_action (self->state, src, type);
            }

        default:
            nn_fsm_bad_source (self->state, src, type);
        }

/******************************************************************************/
/*  NN_SOFI_STATE_ACTIVE state.                                               */
/*  We have an established connection, all events here are regarding the data */
/*  I/O and the one shutdown endpoint event.                                  */
/******************************************************************************/
    case NN_SOFI_STATE_ACTIVE:
        switch (src) {

        /* ========================= */
        /*  Endpoint EQ Action       */
        /* ========================= */
        case NN_SOFI_SRC_ENDPOINT | OFI_SRC_EQ:
            switch (type) {
            case FI_SHUTDOWN:

                _ofi_debug("OFI[S]: Remote endpoint disconnected!\n");
                self->socket_state = NN_SOFI_SOCKET_STATE_CLOSED;

                /* The connection is dropped from the remote end.
                   This is an unrecoverable error and we should terminate */
                nn_sofi_critical_error( self, -EINTR );
                return;

            default:

                /* EQ Error */
                if (type < 0) {

                    /* Unrecoverable comm error */
                    nn_sofi_critical_error( self, type );
                    return;

                }

                nn_fsm_bad_action (self->state, src, type);
            }

        /* ========================= */
        /*  Endpoint RX CQ Event     */
        /* ========================= */
        case NN_SOFI_SRC_ENDPOINT | OFI_SRC_CQ_RX:
            switch (type) {
            case NN_OFIW_COMPLETED:

                /* Get CQ Event */
                cq_entry = (struct fi_cq_msg_entry *) srcptr;

                /* Process incoming data */
                nn_sofi_ingress_handle( self, cq_entry );

                /* Post (if possible) ingress buffers */
                ret = nn_sofi_ingress_post( self );
                if (ret == -EAGAIN) {

                    /* Set postlater flag if not possible to post buffer */
                    _ofi_debug("OFI[S]: Back-pressure from ingress\n");
                    self->ingress_flags |= NN_SOFI_IN_FLAG_POSTLATER;

                }
                return;

            case NN_OFIW_ERROR:

                /* Get CQ Error */
                cq_error = (struct fi_cq_err_entry *) srcptr;

                /* Unrecoverable error while receiving data */
                nn_sofi_ingress_handle_error( self, cq_error );
                nn_sofi_critical_error( self, -cq_error->err );

                return;

            default:
                nn_fsm_bad_action (self->state, src, type);
            }

        /* ========================= */
        /*  Endpoint TX CQ Event     */
        /* ========================= */
        case NN_SOFI_SRC_ENDPOINT | OFI_SRC_CQ_TX:
            switch (type) {
            case NN_OFIW_COMPLETED:

                /* Get CQ Event */
                cq_entry = (struct fi_cq_msg_entry *) srcptr;

                /* Data from the output buffer are sent */
                nn_sofi_egress_handle( self, cq_entry );

                return;

            case NN_OFIW_ERROR:

                /* Get CQ Error */
                cq_error = (struct fi_cq_err_entry *) srcptr;

                /* Unrecoverable error while sending data */
                nn_sofi_egress_handle_error( self, cq_error );
                nn_sofi_critical_error( self, -cq_error->err );
                return;

            default:
                nn_fsm_bad_action (self->state, src, type);
            }

        /* ========================= */
        /*  Keepalive Ticks Timer    */
        /* ========================= */
        case NN_SOFI_SRC_KEEPALIVE_TIMER:
            switch (type) {
            case NN_TIMER_TIMEOUT:

                /* Handhsake phase is completely terminated */
                _ofi_debug("OFI[S]: Keepalive tick\n");

                /* Check if connection timed out */
                if (++self->ticks_in > NN_SOFI_KEEPALIVE_IN_TICKS) {
                    _ofi_debug("OFI[S]: Keepalive expired, dropping"
                        " connection\n");

                    /* Reset and stop timer */
                    self->ticks_in = 0;
                    nn_timer_stop( &self->timer_keepalive );

                    /* Drop connection through a critical error */
                    nn_sofi_critical_error( self, -ETIMEDOUT );
                    return;
                }

                /* Check if we should send a keepalive packet */
                if (++self->ticks_out > NN_SOFI_KEEPALIVE_OUT_TICKS) {
                    self->ticks_out = 0;

                    /* Send aux packet */
                    _ofi_debug("OFI[S]: Sending KEEPALIVE\n");
                    ret = nn_sofi_egress_post_aux( self,
                        NN_SOFI_KEEPALIVE_PACKET_LEN );
                    if (ret) {
                        FT_PRINTERR("nn_sofi_egress_post_aux", ret);
                        nn_sofi_critical_error( self, ret );
                        return;
                    }

                }

                /* Stop Keepalive timer only to be started later */
                nn_timer_stop( &self->timer_keepalive );
                return;

            case NN_TIMER_STOPPED:

                /* Restart Keepalive timer */
                _ofi_debug("OFI[S]: Keepalive stopped, restarting\n");
                nn_timer_start( &self->timer_keepalive,
                    NN_SOFI_TIMEOUT_KEEPALIVE_TICK );
                return;

            default:
                nn_fsm_bad_action (self->state, src, type);
            }

        default:
            nn_fsm_bad_source (self->state, src, type);
        }

/******************************************************************************/
/*  Invalid state.                                                            */
/******************************************************************************/
    default:
        nn_fsm_bad_state (self->state, src, type);

    }

}


text:
// Copyright 2000-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license that can be found in the LICENSE file.
package com.intellij.xdebugger.impl.evaluate;

import com.intellij.ide.lightEdit.LightEdit;
import com.intellij.openapi.diagnostic.Logger;
import com.intellij.openapi.editor.Document;
import com.intellij.openapi.editor.EditorLinePainter;
import com.intellij.openapi.editor.LineExtensionInfo;
import com.intellij.openapi.editor.colors.EditorColorsManager;
import com.intellij.openapi.editor.markup.TextAttributes;
import com.intellij.openapi.fileEditor.FileDocumentManager;
import com.intellij.openapi.project.Project;
import com.intellij.openapi.util.Couple;
import com.intellij.openapi.util.Key;
import com.intellij.openapi.util.NlsSafe;
import com.intellij.openapi.util.registry.Registry;
import com.intellij.openapi.util.text.StringUtil;
import com.intellij.openapi.vfs.VirtualFile;
import com.intellij.ui.Gray;
import com.intellij.ui.JBColor;
import com.intellij.ui.SimpleColoredText;
import com.intellij.ui.SimpleTextAttributes;
import com.intellij.util.containers.ContainerUtil;
import com.intellij.xdebugger.XDebugSession;
import com.intellij.xdebugger.XDebuggerManager;
import com.intellij.xdebugger.XSourcePosition;
import com.intellij.xdebugger.frame.presentation.XValuePresentation;
import com.intellij.xdebugger.impl.XDebugSessionImpl;
import com.intellij.xdebugger.impl.XDebuggerManagerImpl;
import com.intellij.xdebugger.impl.frame.XVariablesView;
import com.intellij.xdebugger.impl.inline.XDebuggerInlayUtil;
import com.intellij.xdebugger.impl.ui.tree.nodes.XValueNodeImpl;
import com.intellij.xdebugger.impl.ui.tree.nodes.XValueTextRendererImpl;
import com.intellij.xdebugger.settings.XDebuggerSettingsManager;
import com.intellij.xdebugger.ui.DebuggerColors;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

import java.awt.*;
import java.util.List;
import java.util.*;

/**
 * @author Konstantin Bulenkov
 */
public class XDebuggerEditorLinePainter extends EditorLinePainter {
  private static final Logger LOG = Logger.getInstance(XDebuggerEditorLinePainter.class);
  public static final Key<Map<Variable, VariableValue>> CACHE = Key.create("debug.inline.variables.cache");
  // we want to limit number of line extensions to avoid very slow painting
  // the constant is rather random (feel free to adjust it upon getting a new information)
  private static final int LINE_EXTENSIONS_MAX_COUNT = 200;

  @Override
  public Collection<LineExtensionInfo> getLineExtensions(@NotNull Project project, @NotNull VirtualFile file, int lineNumber) {
    if (LightEdit.owns(project)) return null;
    if (!XDebuggerSettingsManager.getInstance().getDataViewSettings().isShowValuesInline()) {
      return null;
    }

    XDebugSession session = XDebuggerManager.getInstance(project).getCurrentSession();
    if (session == null) {
      return null;
    }

    XVariablesView.InlineVariablesInfo data = XVariablesView.InlineVariablesInfo.get(session);
    final Document doc = FileDocumentManager.getInstance().getDocument(file);

    if (data == null || doc == null) {
      return null;
    }

    Map<Variable, VariableValue> oldValues = getOldValues(project);
    List<XValueNodeImpl> values = data.get(file, lineNumber, doc.getModificationStamp());
    if (values != null && !values.isEmpty()) {
      final TextAttributes attributes = getAttributes(lineNumber, file, session);

      ArrayList<VariableText> result = new ArrayList<>();
      for (XValueNodeImpl value : values) {
        SimpleColoredText text = createPresentation(value);
        if (text == null) continue;

        final String name = value.getName();
        if (StringUtil.isEmpty(text.toString())) {
          continue;
        }

        final VariableText res = computeVariablePresentationWithChanges(value, name, text, attributes, lineNumber, oldValues);
        result.add(res);
      }
      final List<LineExtensionInfo> infos = new ArrayList<>();
      for (VariableText text : result) {
        LineExtensionInfo varNameInfo = text.infos.get(0);
        LineExtensionInfo wrappedName =
          new LineExtensionInfo("  " + varNameInfo.getText() + XDebuggerInlayUtil.INLINE_HINTS_DELIMETER + " ",
                                varNameInfo.getColor(),
                                varNameInfo.getEffectType(),
                                varNameInfo.getEffectColor(),
                                varNameInfo.getFontType());
        List<LineExtensionInfo> value = text.infos.subList(1, text.infos.size());
        infos.add(wrappedName);
        infos.addAll(value);
      }
      return ContainerUtil.getFirstItems(infos, LINE_EXTENSIONS_MAX_COUNT);
    }
    return null;
  }

  public static SimpleColoredText computeVariablePresentationWithChanges(XValueNodeImpl value,
                                                                         String name,
                                                                         SimpleColoredText text,
                                                                         TextAttributes attributes,
                                                                         int lineNumber,
                                                                         Project project) {
    Map<Variable, VariableValue> oldValues = getOldValues(project);
    VariableText variableText = computeVariablePresentationWithChanges(value, name, text, attributes, lineNumber, oldValues);

    SimpleColoredText coloredText = new SimpleColoredText();
    for (LineExtensionInfo info : variableText.infos) {
      TextAttributes textAttributes = new TextAttributes(info.getColor(), info.getBgColor(), info.getEffectColor(),info.getEffectType(), info.getFontType());
      coloredText.append(info.getText(), SimpleTextAttributes.fromTextAttributes(textAttributes));
    }
    return coloredText;
  }

  @NotNull
  private static VariableText computeVariablePresentationWithChanges(XValueNodeImpl value,
                                                                     @NlsSafe String name,
                                                                     SimpleColoredText text,
                                                                     TextAttributes attributes,
                                                                     int lineNumber,
                                                                     Map<Variable, VariableValue> oldValues) {
    final VariableText res = new VariableText();
    res.add(new LineExtensionInfo(name, attributes));

    Variable var = new Variable(name, lineNumber);
    VariableValue variableValue = oldValues.computeIfAbsent(var, k -> new VariableValue(text.toString(), null, value.hashCode()));
    if (variableValue.valueNodeHashCode != value.hashCode()) {
      variableValue.old = variableValue.actual;
      variableValue.actual = text.toString();
      variableValue.valueNodeHashCode = value.hashCode();
    }

    if (!variableValue.isChanged()) {
      ArrayList<String> texts = text.getTexts();
      for (int i = 0; i < texts.size(); i++) {
        String s = texts.get(i);
        TextAttributes attr = Registry.is("debugger.show.values.colorful")
                              ? text.getAttributes().get(i).toTextAttributes()
                              : attributes;
        res.add(new LineExtensionInfo(s, attr));
      }
    }
    else {
      variableValue.produceChangedParts(res.infos);
    }
    return res;
  }

  @NotNull
  public static Map<Variable, VariableValue> getOldValues(@NotNull Project project) {
    Map<Variable, VariableValue> oldValues = project.getUserData(CACHE);
    if (oldValues == null) {
      oldValues = new HashMap<>();
      project.putUserData(CACHE, oldValues);
    }
    return oldValues;
  }

  @NotNull
  public static TextAttributes getAttributes(int lineNumber, @NotNull VirtualFile file, XDebugSession session) {
    final int bpLine = getCurrentBreakPointLineInFile(session, file);
    boolean isTopFrame = session instanceof XDebugSessionImpl && ((XDebugSessionImpl)session).isTopFrameSelected();
    return bpLine == lineNumber
           && isTopFrame
           && ((XDebuggerManagerImpl)XDebuggerManager.getInstance(session.getProject())).isFullLineHighlighter()
           ? getTopFrameSelectedAttributes() : getNormalAttributes();
  }

  @Nullable
  public static SimpleColoredText createPresentation(@NotNull XValueNodeImpl value) {
    SimpleColoredText text = new SimpleColoredText();
    XValueTextRendererImpl renderer = new XValueTextRendererImpl(text);
    final XValuePresentation presentation = value.getValuePresentation();
    if (presentation == null) return null;
    try {
      if (presentation instanceof XValueCompactPresentation && !value.getTree().isUnderRemoteDebug()) {
        ((XValueCompactPresentation)presentation).renderValue(renderer, value);
      }
      else {
        presentation.renderValue(renderer);
      }
      if (StringUtil.isEmpty(text.toString())) {
        final String type = value.getValuePresentation().getType();
        if (!StringUtil.isEmpty(type)) {
          text.append(type, SimpleTextAttributes.REGULAR_ATTRIBUTES);
        }
      }
    }
    catch (Exception e) {
      LOG.error(e);
      return null;
    }
    return text;
  }

  private static int getCurrentBreakPointLineInFile(@Nullable XDebugSession session, VirtualFile file) {
    try {
      if (session != null) {
        final XSourcePosition position = session.getCurrentPosition();
        if (position != null && position.getFile().equals(file)) {
          return position.getLine();
        }
      }
    } catch (Exception ignore){}
    return -1;
  }

  private static TextAttributes getNormalAttributes() {
    TextAttributes attributes = EditorColorsManager.getInstance().getGlobalScheme().getAttributes(DebuggerColors.INLINED_VALUES);
    if (attributes == null || attributes.getForegroundColor() == null) {
     return new TextAttributes(new JBColor(() -> EditorColorsManager.getInstance().isDarkEditor() ? new Color(0x3d8065) : Gray._135), null, null, null, Font.ITALIC);
    }
    return attributes;
  }

  private static TextAttributes getChangedAttributes() {
    TextAttributes attributes = EditorColorsManager.getInstance().getGlobalScheme().getAttributes(DebuggerColors.INLINED_VALUES_MODIFIED);
    if (attributes == null || attributes.getForegroundColor() == null) {
      return new TextAttributes(new JBColor(() -> EditorColorsManager.getInstance().isDarkEditor() ? new Color(0xa1830a) : new Color(0xca8021)), null, null, null, Font.ITALIC);
    }
    return attributes;
  }

  private static TextAttributes getTopFrameSelectedAttributes() {
    TextAttributes attributes = EditorColorsManager.getInstance().getGlobalScheme().getAttributes(DebuggerColors.INLINED_VALUES_EXECUTION_LINE);
    if (attributes == null || attributes.getForegroundColor() == null) {
      //noinspection UseJBColor
      return new TextAttributes(EditorColorsManager.getInstance().isDarkEditor() ? new Color(255, 235, 9) : new Color(0, 255, 86), null, null, null, Font.ITALIC);
    }
    return attributes;
  }

  static class Variable {
    private final int lineNumber;
    private final String name;

    Variable(String name, int lineNumber) {
      this.lineNumber = lineNumber;
      this.name = name;
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) return true;
      if (o == null || getClass() != o.getClass()) return false;

      Variable variable = (Variable)o;

      if (lineNumber != variable.lineNumber) return false;
      if (!name.equals(variable.name)) return false;

      return true;
    }

    @Override
    public int hashCode() {
      return Objects.hash(lineNumber, name);
    }
  }

  static class VariableValue {
    // TODO: this have to be specified somewhere in XValuePresentation
    private static final List<Couple<String>> ARRAYS_WRAPPERS = List.of(Couple.of("[", "]"), Couple.of("{", "}"));
    private static final String ARRAY_DELIMITER = ", ";
    private @NlsSafe String actual;
    private @NlsSafe String old;
    private int valueNodeHashCode;

    VariableValue(String actual, String old, int valueNodeHashCode) {
      this.actual = actual;
      this.old = old;
      this.valueNodeHashCode = valueNodeHashCode;
    }

    public boolean isChanged() {
      return old != null && !StringUtil.equals(actual, old);
    }

    void produceChangedParts(List<? super LineExtensionInfo> result) {
      Couple<String> wrapperActual = getArrayWrapper(actual);
      if (wrapperActual != null && getArrayWrapper(old) != null) {
        List<String> actualParts = getArrayParts(actual);
        List<String> oldParts = getArrayParts(old);
        //noinspection HardCodedStringLiteral
        result.add(new LineExtensionInfo(wrapperActual.first, getNormalAttributes()));
        for (int i = 0; i < actualParts.size(); i++) {
          if (i < oldParts.size() && StringUtil.equals(actualParts.get(i), oldParts.get(i))) {
            result.add(new LineExtensionInfo(actualParts.get(i), getNormalAttributes()));
          } else {
            result.add(new LineExtensionInfo(actualParts.get(i), getChangedAttributes()));
          }
          if (i != actualParts.size() - 1) {
            result.add(new LineExtensionInfo(ARRAY_DELIMITER, getNormalAttributes()));
          }
        }
        //noinspection HardCodedStringLiteral
        result.add(new LineExtensionInfo(wrapperActual.second, getNormalAttributes()));
        return;
      }

      result.add(new LineExtensionInfo(actual, getChangedAttributes()));
    }

    @Nullable
    private static Couple<String> getArrayWrapper(@Nullable String s) {
      if (s == null) {
        return null;
      }
      return ContainerUtil.find(ARRAYS_WRAPPERS, w -> s.startsWith(w.first) && s.endsWith(w.second));
    }

    private static List<String> getArrayParts(String array) {
      return StringUtil.split(array.substring(1, array.length() - 1), ARRAY_DELIMITER);
    }
  }

  private static class VariableText {
    final List<LineExtensionInfo> infos = new ArrayList<>();

    void add(LineExtensionInfo info) {
      infos.add(info);
    }
  }
}

redpajama_set_name: RedPajamaGithub


text:
package main

import (
	"bufio"
	"fmt"
	"golang.org/x/net/websocket"
	"html/template"
	"io"
	"net/http"
	"os"
	"strconv"
	"time"
)

var address = "0.0.0.0"
var port = 1234
var gws *websocket.Conn

func echo(ws *websocket.Conn) {
	gws = ws
	for {
		var msg string
		if err := websocket.Message.Receive(ws, &msg); err != nil {
			fmt.Println("can't receive" + err.Error())
			break
		}
		fmt.Println("Receive back from client: " + msg)
	}
}

func tail(in io.Reader) {
	buf := bufio.NewReader(in)

	for {
		if gws != nil {
			break
		}
		time.Sleep(10 * time.Millisecond)
	}

	for {
		line, _, err := buf.ReadLine()
		if len(line) > 0 {
			if err = websocket.Message.Send(gws, string(line)); err != nil {
				fmt.Println("Cant send")
				break
			}
		}
		if err != nil {
			if err != io.EOF {
				fmt.Fprintln(os.Stderr, err.Error())
				os.Exit(1)
			}
			time.Sleep(10 * time.Millisecond)
		}
	}
}

func bootTailingServer(port int) {
	var in io.Reader
	if len(os.Args) > 1 {
		fin, err := os.Open(os.Args[1])
		if err != nil {
			fmt.Fprintln(os.Stderr, err)
			os.Exit(1)
		}
		defer fin.Close()
		in = fin
	} else {
		in = os.Stdin
	}

	http.Handle("/log", websocket.Handler(echo))
	http.HandleFunc("/", viewHandler)

	go tail(in)

	err := http.ListenAndServe(":"+strconv.Itoa(port), nil)
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

func viewHandler(w http.ResponseWriter, r *http.Request) {
	t := template.Must(parseAssets("assets/tailer.html"))
	err := t.Execute(w, struct {
		Title string
		Address string
		Port  int
	}{Title: os.Args[1], Address: address, Port: port})
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
	}
}

func parseAssets(path string) (*template.Template, error) {
	src, err := Asset(path)
	if err != nil {
		return nil, err
	}
	return template.New(path).Parse(string(src))
}

func main() {
	// Boot tailing Server
	bootTailingServer(port)
	fmt.Println(fmt.Sprintf("Open localhost:%d on browser", port))
}


text:
package com.pushinginertia.commons.core.validation;

import org.junit.Assert;
import org.junit.Test;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

public class ValidateAsTest {
	@Test(expected = IllegalArgumentException.class)
	public void notNull() {
		ValidateAs.notNull(null, "nullParameter");
	}

	@Test
	public void notEmpty() {
		ValidateAs.notEmpty("abc", "emptyParameter");
	}

	@Test(expected = IllegalArgumentException.class)
	public void notEmptyFail() {
		ValidateAs.notEmpty("", "emptyParameter");
	}

	@Test
	public void trimmedNotEmpty() {
		Assert.assertEquals("abc", ValidateAs.trimmedNotEmpty(" abc ", "emptyParameter"));
	}

	@Test(expected = IllegalArgumentException.class)
	public void trimmedNotEmptyFail() {
		ValidateAs.trimmedNotEmpty("  ", "emptyParameter");
	}

	@Test
	public void notEqual() {
		ValidateAs.notEqual("abc", "def", "string1", "string2");
	}

	@Test(expected = IllegalArgumentException.class)
	public void notEqualFail() {
		ValidateAs.notEqual("abc", "abc", "string1", "string2");
	}

	@Test
	public void equal() {
		ValidateAs.equal("abc", "abc", "string1", "string2");
	}

	@Test(expected = IllegalArgumentException.class)
	public void equalFail() {
		ValidateAs.equal("abc", "def", "string1", "string2");
	}

	@Test(expected = IllegalArgumentException.class)
	public void positive0() {
		ValidateAs.positive(0, "zeroParameter");
	}

	@Test(expected = IllegalArgumentException.class)
	public void positiveMinus1() {
		ValidateAs.positive(-1, "minusOneParameter");
	}

	@Test
	public void nonNegative() {
		ValidateAs.nonNegative(BigDecimal.ZERO, "bd1");
		ValidateAs.nonNegative(BigDecimal.ONE, "bd2");
	}

	@Test(expected = IllegalArgumentException.class)
	public void nonNegativeFail() {
		ValidateAs.nonNegative(new BigDecimal(-1), "bd1");
	}

	@Test
	public void mapDoesNotContainKey() {
		final Map<String, String> m = new HashMap<String, String>();

		// positive case
		ValidateAs.mapDoesNotContainKey(m, "a");

		// negative case
		m.put("a", "1");
		try {
			ValidateAs.mapDoesNotContainKey(m, "a");
			Assert.fail("Expected IllegalArgumentException");
		} catch (final IllegalArgumentException e) {
			// expected
		}
	}

	@Test
	public void ofLength() {
		ValidateAs.ofLength("abc", 3, "string1");
	}

	@Test(expected = IllegalArgumentException.class)
	public void ofLengthFail() {
		ValidateAs.ofLength("abc", 2, "string1");
	}

	@Test
	public void ofMinimumLength() {
		ValidateAs.ofMinimumLength("abc", 3, "string1");
		ValidateAs.ofMinimumLength("abcd", 3, "string1");
	}

	@Test(expected = IllegalArgumentException.class)
	public void ofMinimumLengthFail() {
		ValidateAs.ofMinimumLength("ab", 3, "string1");
	}

	@Test
	public void allUppercase() {
		ValidateAs.allUppercase("ABC", "string");
	}

	@Test(expected = IllegalArgumentException.class)
	public void allUppercaseFail1() {
		ValidateAs.allUppercase("ABc", "string");
	}

	@Test(expected = IllegalArgumentException.class)
	public void allUppercaseFail2() {
		ValidateAs.allUppercase("ABC1", "string");
	}

	@Test
	public void indexInList() {
		ValidateAs.indexInList(Arrays.asList(1, 2, 3), 0, "listname");
		ValidateAs.indexInList(Arrays.asList(1, 2, 3), 1, "listname");
		ValidateAs.indexInList(Arrays.asList(1, 2, 3), 2, "listname");
	}

	@Test(expected = IllegalArgumentException.class)
	public void indexInListFailNegativeIndex() {
		ValidateAs.indexInList(Arrays.asList(1, 2, 3), -1, "listname");
	}

	@Test(expected = IllegalArgumentException.class)
	public void indexInListFailOutOfBoundsIndex() {
		ValidateAs.indexInList(Arrays.asList(1, 2, 3), 3, "listname");
	}

	@Test(expected = IllegalArgumentException.class)
	public void indexInListFailEmptyList() {
		ValidateAs.indexInList(new ArrayList(), 0, "listname");
	}

	@Test(expected = IllegalArgumentException.class)
	public void indexInListFailNullList() {
		ValidateAs.indexInList(null, 0, "listname");
	}
}


text:
package com.intellij.cvsSupport2.cvshandlers;

import com.intellij.cvsSupport2.cvsoperations.cvsMessages.CvsMessagesListener;
import com.intellij.cvsSupport2.cvsoperations.cvsMessages.CvsCompositeListener;
import com.intellij.cvsSupport2.cvsoperations.cvsMessages.CvsCompositeListener;
import com.intellij.cvsSupport2.cvsoperations.cvsMessages.CvsMessagesListener;

/**
 * author: lesya
 */
public abstract class AbstractCvsHandler extends CvsHandler {
  protected final CvsCompositeListener myCompositeListener = new CvsCompositeListener();

  public AbstractCvsHandler(String title, FileSetToBeUpdated files) {
    super(title, files);
  }

  public void addCvsListener(CvsMessagesListener listener) {
    myCompositeListener.addCvsListener(listener);
  }

  public void removeCvsListener(CvsMessagesListener listener) {
    myCompositeListener.removeCvsListener(listener);
  }
}


text:
/*
 * VLAN		An implementation of 802.1Q VLAN tagging.
 *
 * Authors:	Ben Greear <greearb@candelatech.com>
 *
 *		This program is free software; you can redistribute it and/or
 *		modify it under the terms of the GNU General Public License
 *		as published by the Free Software Foundation; either version
 *		2 of the License, or (at your option) any later version.
 *
 */
#ifndef _LINUX_IF_VLAN_H_
#define _LINUX_IF_VLAN_H_

#include <linux/netdevice.h>
#include <linux/etherdevice.h>
#include <linux/rtnetlink.h>
#include <linux/bug.h>
#include <uapi/linux/if_vlan.h>

#define VLAN_HLEN	4		/* The additional bytes required by VLAN
					 * (in addition to the Ethernet header)
					 */
#define VLAN_ETH_HLEN	18		/* Total octets in header.	 */
#define VLAN_ETH_ZLEN	64		/* Min. octets in frame sans FCS */

/*
 * According to 802.3ac, the packet can be 4 bytes longer. --Klika Jan
 */
#define VLAN_ETH_DATA_LEN	1500	/* Max. octets in payload	 */
#define VLAN_ETH_FRAME_LEN	1518	/* Max. octets in frame sans FCS */

/*
 * 	struct vlan_hdr - vlan header
 * 	@h_vlan_TCI: priority and VLAN ID
 *	@h_vlan_encapsulated_proto: packet type ID or len
 */
struct vlan_hdr {
	__be16	h_vlan_TCI;
	__be16	h_vlan_encapsulated_proto;
};

/**
 *	struct vlan_ethhdr - vlan ethernet header (ethhdr + vlan_hdr)
 *	@h_dest: destination ethernet address
 *	@h_source: source ethernet address
 *	@h_vlan_proto: ethernet protocol
 *	@h_vlan_TCI: priority and VLAN ID
 *	@h_vlan_encapsulated_proto: packet type ID or len
 */
struct vlan_ethhdr {
	unsigned char	h_dest[ETH_ALEN];
	unsigned char	h_source[ETH_ALEN];
	__be16		h_vlan_proto;
	__be16		h_vlan_TCI;
	__be16		h_vlan_encapsulated_proto;
};

#include <linux/skbuff.h>

static inline struct vlan_ethhdr *vlan_eth_hdr(const struct sk_buff *skb)
{
	return (struct vlan_ethhdr *)skb_mac_header(skb);
}

#define VLAN_PRIO_MASK		0xe000 /* Priority Code Point */
#define VLAN_PRIO_SHIFT		13
#define VLAN_CFI_MASK		0x1000 /* Canonical Format Indicator */
#define VLAN_TAG_PRESENT	VLAN_CFI_MASK
#define VLAN_VID_MASK		0x0fff /* VLAN Identifier */
#define VLAN_N_VID		4096

/* found in socket.c */
extern void vlan_ioctl_set(int (*hook)(struct net *, void __user *));

static inline bool is_vlan_dev(struct net_device *dev)
{
        return dev->priv_flags & IFF_802_1Q_VLAN;
}

#define skb_vlan_tag_present(__skb)	((__skb)->vlan_tci & VLAN_TAG_PRESENT)
#define skb_vlan_tag_get(__skb)		((__skb)->vlan_tci & ~VLAN_TAG_PRESENT)
#define skb_vlan_tag_get_id(__skb)	((__skb)->vlan_tci & VLAN_VID_MASK)

/**
 *	struct vlan_pcpu_stats - VLAN percpu rx/tx stats
 *	@rx_packets: number of received packets
 *	@rx_bytes: number of received bytes
 *	@rx_multicast: number of received multicast packets
 *	@tx_packets: number of transmitted packets
 *	@tx_bytes: number of transmitted bytes
 *	@syncp: synchronization point for 64bit counters
 *	@rx_errors: number of rx errors
 *	@tx_dropped: number of tx drops
 */
struct vlan_pcpu_stats {
	u64			rx_packets;
	u64			rx_bytes;
	u64			rx_multicast;
	u64			tx_packets;
	u64			tx_bytes;
	struct u64_stats_sync	syncp;
	u32			rx_errors;
	u32			tx_dropped;
};

#if defined(CONFIG_VLAN_8021Q) || defined(CONFIG_VLAN_8021Q_MODULE)

extern struct net_device *__vlan_find_dev_deep_rcu(struct net_device *real_dev,
					       __be16 vlan_proto, u16 vlan_id);
extern struct net_device *vlan_dev_real_dev(const struct net_device *dev);
extern u16 vlan_dev_vlan_id(const struct net_device *dev);
extern __be16 vlan_dev_vlan_proto(const struct net_device *dev);

/**
 *	struct vlan_priority_tci_mapping - vlan egress priority mappings
 *	@priority: skb priority
 *	@vlan_qos: vlan priority: (skb->priority << 13) & 0xE000
 *	@next: pointer to next struct
 */
struct vlan_priority_tci_mapping {
	u32					priority;
	u16					vlan_qos;
	struct vlan_priority_tci_mapping	*next;
};

struct proc_dir_entry;
struct netpoll;

/**
 *	struct vlan_dev_priv - VLAN private device data
 *	@nr_ingress_mappings: number of ingress priority mappings
 *	@ingress_priority_map: ingress priority mappings
 *	@nr_egress_mappings: number of egress priority mappings
 *	@egress_priority_map: hash of egress priority mappings
 *	@vlan_proto: VLAN encapsulation protocol
 *	@vlan_id: VLAN identifier
 *	@flags: device flags
 *	@real_dev: underlying netdevice
 *	@real_dev_addr: address of underlying netdevice
 *	@dent: proc dir entry
 *	@vlan_pcpu_stats: ptr to percpu rx stats
 */
struct vlan_dev_priv {
	unsigned int				nr_ingress_mappings;
	u32					ingress_priority_map[8];
	unsigned int				nr_egress_mappings;
	struct vlan_priority_tci_mapping	*egress_priority_map[16];

	__be16					vlan_proto;
	u16					vlan_id;
	u16					flags;

	struct net_device			*real_dev;
	unsigned char				real_dev_addr[ETH_ALEN];

	struct proc_dir_entry			*dent;
	struct vlan_pcpu_stats __percpu		*vlan_pcpu_stats;
#ifdef CONFIG_NET_POLL_CONTROLLER
	struct netpoll				*netpoll;
#endif
	unsigned int				nest_level;
};

static inline struct vlan_dev_priv *vlan_dev_priv(const struct net_device *dev)
{
	return netdev_priv(dev);
}

static inline u16
vlan_dev_get_egress_qos_mask(struct net_device *dev, u32 skprio)
{
	struct vlan_priority_tci_mapping *mp;

	smp_rmb(); /* coupled with smp_wmb() in vlan_dev_set_egress_priority() */

	mp = vlan_dev_priv(dev)->egress_priority_map[(skprio & 0xF)];
	while (mp) {
		if (mp->priority == skprio) {
			return mp->vlan_qos; /* This should already be shifted
					      * to mask correctly with the
					      * VLAN's TCI */
		}
		mp = mp->next;
	}
	return 0;
}

extern bool vlan_do_receive(struct sk_buff **skb);

extern int vlan_vid_add(struct net_device *dev, __be16 proto, u16 vid);
extern void vlan_vid_del(struct net_device *dev, __be16 proto, u16 vid);

extern int vlan_vids_add_by_dev(struct net_device *dev,
				const struct net_device *by_dev);
extern void vlan_vids_del_by_dev(struct net_device *dev,
				 const struct net_device *by_dev);

extern bool vlan_uses_dev(const struct net_device *dev);

static inline int vlan_get_encap_level(struct net_device *dev)
{
	BUG_ON(!is_vlan_dev(dev));
	return vlan_dev_priv(dev)->nest_level;
}
#else
static inline struct net_device *
__vlan_find_dev_deep_rcu(struct net_device *real_dev,
		     __be16 vlan_proto, u16 vlan_id)
{
	return NULL;
}

static inline struct net_device *vlan_dev_real_dev(const struct net_device *dev)
{
	BUG();
	return NULL;
}

static inline u16 vlan_dev_vlan_id(const struct net_device *dev)
{
	BUG();
	return 0;
}

static inline __be16 vlan_dev_vlan_proto(const struct net_device *dev)
{
	BUG();
	return 0;
}

static inline u16 vlan_dev_get_egress_qos_mask(struct net_device *dev,
					       u32 skprio)
{
	return 0;
}

static inline bool vlan_do_receive(struct sk_buff **skb)
{
	return false;
}

static inline int vlan_vid_add(struct net_device *dev, __be16 proto, u16 vid)
{
	return 0;
}

static inline void vlan_vid_del(struct net_device *dev, __be16 proto, u16 vid)
{
}

static inline int vlan_vids_add_by_dev(struct net_device *dev,
				       const struct net_device *by_dev)
{
	return 0;
}

static inline void vlan_vids_del_by_dev(struct net_device *dev,
					const struct net_device *by_dev)
{
}

static inline bool vlan_uses_dev(const struct net_device *dev)
{
	return false;
}
static inline int vlan_get_encap_level(struct net_device *dev)
{
	BUG();
	return 0;
}
#endif

static inline bool vlan_hw_offload_capable(netdev_features_t features,
					   __be16 proto)
{
	if (proto == htons(ETH_P_8021Q) && features & NETIF_F_HW_VLAN_CTAG_TX)
		return true;
	if (proto == htons(ETH_P_8021AD) && features & NETIF_F_HW_VLAN_STAG_TX)
		return true;
	return false;
}

/**
 * __vlan_insert_tag - regular VLAN tag inserting
 * @skb: skbuff to tag
 * @vlan_proto: VLAN encapsulation protocol
 * @vlan_tci: VLAN TCI to insert
 *
 * Inserts the VLAN tag into @skb as part of the payload
 * Returns error if skb_cow_head failes.
 *
 * Does not change skb->protocol so this function can be used during receive.
 */
static inline int __vlan_insert_tag(struct sk_buff *skb,
				    __be16 vlan_proto, u16 vlan_tci)
{
	struct vlan_ethhdr *veth;

	if (skb_cow_head(skb, VLAN_HLEN) < 0)
		return -ENOMEM;

	veth = (struct vlan_ethhdr *)skb_push(skb, VLAN_HLEN);

	/* Move the mac addresses to the beginning of the new header. */
	memmove(skb->data, skb->data + VLAN_HLEN, 2 * ETH_ALEN);
	skb->mac_header -= VLAN_HLEN;

	/* first, the ethernet type */
	veth->h_vlan_proto = vlan_proto;

	/* now, the TCI */
	veth->h_vlan_TCI = htons(vlan_tci);

	return 0;
}

/**
 * vlan_insert_tag - regular VLAN tag inserting
 * @skb: skbuff to tag
 * @vlan_proto: VLAN encapsulation protocol
 * @vlan_tci: VLAN TCI to insert
 *
 * Inserts the VLAN tag into @skb as part of the payload
 * Returns a VLAN tagged skb. If a new skb is created, @skb is freed.
 *
 * Following the skb_unshare() example, in case of error, the calling function
 * doesn't have to worry about freeing the original skb.
 *
 * Does not change skb->protocol so this function can be used during receive.
 */
static inline struct sk_buff *vlan_insert_tag(struct sk_buff *skb,
					      __be16 vlan_proto, u16 vlan_tci)
{
	int err;

	err = __vlan_insert_tag(skb, vlan_proto, vlan_tci);
	if (err) {
		dev_kfree_skb_any(skb);
		return NULL;
	}
	return skb;
}

/**
 * vlan_insert_tag_set_proto - regular VLAN tag inserting
 * @skb: skbuff to tag
 * @vlan_proto: VLAN encapsulation protocol
 * @vlan_tci: VLAN TCI to insert
 *
 * Inserts the VLAN tag into @skb as part of the payload
 * Returns a VLAN tagged skb. If a new skb is created, @skb is freed.
 *
 * Following the skb_unshare() example, in case of error, the calling function
 * doesn't have to worry about freeing the original skb.
 */
static inline struct sk_buff *vlan_insert_tag_set_proto(struct sk_buff *skb,
							__be16 vlan_proto,
							u16 vlan_tci)
{
	skb = vlan_insert_tag(skb, vlan_proto, vlan_tci);
	if (skb)
		skb->protocol = vlan_proto;
	return skb;
}

/*
 * __vlan_hwaccel_push_inside - pushes vlan tag to the payload
 * @skb: skbuff to tag
 *
 * Pushes the VLAN tag from @skb->vlan_tci inside to the payload.
 *
 * Following the skb_unshare() example, in case of error, the calling function
 * doesn't have to worry about freeing the original skb.
 */
static inline struct sk_buff *__vlan_hwaccel_push_inside(struct sk_buff *skb)
{
	skb = vlan_insert_tag_set_proto(skb, skb->vlan_proto,
					skb_vlan_tag_get(skb));
	if (likely(skb))
		skb->vlan_tci = 0;
	return skb;
}
/*
 * vlan_hwaccel_push_inside - pushes vlan tag to the payload
 * @skb: skbuff to tag
 *
 * Checks is tag is present in @skb->vlan_tci and if it is, it pushes the
 * VLAN tag from @skb->vlan_tci inside to the payload.
 *
 * Following the skb_unshare() example, in case of error, the calling function
 * doesn't have to worry about freeing the original skb.
 */
static inline struct sk_buff *vlan_hwaccel_push_inside(struct sk_buff *skb)
{
	if (skb_vlan_tag_present(skb))
		skb = __vlan_hwaccel_push_inside(skb);
	return skb;
}

/**
 * __vlan_hwaccel_put_tag - hardware accelerated VLAN inserting
 * @skb: skbuff to tag
 * @vlan_proto: VLAN encapsulation protocol
 * @vlan_tci: VLAN TCI to insert
 *
 * Puts the VLAN TCI in @skb->vlan_tci and lets the device do the rest
 */
static inline void __vlan_hwaccel_put_tag(struct sk_buff *skb,
					  __be16 vlan_proto, u16 vlan_tci)
{
	skb->vlan_proto = vlan_proto;
	skb->vlan_tci = VLAN_TAG_PRESENT | vlan_tci;
}

/**
 * __vlan_get_tag - get the VLAN ID that is part of the payload
 * @skb: skbuff to query
 * @vlan_tci: buffer to store value
 *
 * Returns error if the skb is not of VLAN type
 */
static inline int __vlan_get_tag(const struct sk_buff *skb, u16 *vlan_tci)
{
	struct vlan_ethhdr *veth = (struct vlan_ethhdr *)skb->data;

	if (veth->h_vlan_proto != htons(ETH_P_8021Q) &&
	    veth->h_vlan_proto != htons(ETH_P_8021AD))
		return -EINVAL;

	*vlan_tci = ntohs(veth->h_vlan_TCI);
	return 0;
}

/**
 * __vlan_hwaccel_get_tag - get the VLAN ID that is in @skb->cb[]
 * @skb: skbuff to query
 * @vlan_tci: buffer to store value
 *
 * Returns error if @skb->vlan_tci is not set correctly
 */
static inline int __vlan_hwaccel_get_tag(const struct sk_buff *skb,
					 u16 *vlan_tci)
{
	if (skb_vlan_tag_present(skb)) {
		*vlan_tci = skb_vlan_tag_get(skb);
		return 0;
	} else {
		*vlan_tci = 0;
		return -EINVAL;
	}
}

#define HAVE_VLAN_GET_TAG

/**
 * vlan_get_tag - get the VLAN ID from the skb
 * @skb: skbuff to query
 * @vlan_tci: buffer to store value
 *
 * Returns error if the skb is not VLAN tagged
 */
static inline int vlan_get_tag(const struct sk_buff *skb, u16 *vlan_tci)
{
	if (skb->dev->features & NETIF_F_HW_VLAN_CTAG_TX) {
		return __vlan_hwaccel_get_tag(skb, vlan_tci);
	} else {
		return __vlan_get_tag(skb, vlan_tci);
	}
}

/**
 * vlan_get_protocol - get protocol EtherType.
 * @skb: skbuff to query
 * @type: first vlan protocol
 * @depth: buffer to store length of eth and vlan tags in bytes
 *
 * Returns the EtherType of the packet, regardless of whether it is
 * vlan encapsulated (normal or hardware accelerated) or not.
 */
static inline __be16 __vlan_get_protocol(struct sk_buff *skb, __be16 type,
					 int *depth)
{
	unsigned int vlan_depth = skb->mac_len;

	/* if type is 802.1Q/AD then the header should already be
	 * present at mac_len - VLAN_HLEN (if mac_len > 0), or at
	 * ETH_HLEN otherwise
	 */
	if (type == htons(ETH_P_8021Q) || type == htons(ETH_P_8021AD)) {
		if (vlan_depth) {
			if (WARN_ON(vlan_depth < VLAN_HLEN))
				return 0;
			vlan_depth -= VLAN_HLEN;
		} else {
			vlan_depth = ETH_HLEN;
		}
		do {
			struct vlan_hdr *vh;

			if (unlikely(!pskb_may_pull(skb,
						    vlan_depth + VLAN_HLEN)))
				return 0;

			vh = (struct vlan_hdr *)(skb->data + vlan_depth);
			type = vh->h_vlan_encapsulated_proto;
			vlan_depth += VLAN_HLEN;
		} while (type == htons(ETH_P_8021Q) ||
			 type == htons(ETH_P_8021AD));
	}

	if (depth)
		*depth = vlan_depth;

	return type;
}

/**
 * vlan_get_protocol - get protocol EtherType.
 * @skb: skbuff to query
 *
 * Returns the EtherType of the packet, regardless of whether it is
 * vlan encapsulated (normal or hardware accelerated) or not.
 */
static inline __be16 vlan_get_protocol(struct sk_buff *skb)
{
	return __vlan_get_protocol(skb, skb->protocol, NULL);
}

static inline void vlan_set_encap_proto(struct sk_buff *skb,
					struct vlan_hdr *vhdr)
{
	__be16 proto;
	unsigned short *rawp;

	/*
	 * Was a VLAN packet, grab the encapsulated protocol, which the layer
	 * three protocols care about.
	 */

	proto = vhdr->h_vlan_encapsulated_proto;
	if (eth_proto_is_802_3(proto)) {
		skb->protocol = proto;
		return;
	}

	rawp = (unsigned short *)(vhdr + 1);
	if (*rawp == 0xFFFF)
		/*
		 * This is a magic hack to spot IPX packets. Older Novell
		 * breaks the protocol design and runs IPX over 802.3 without
		 * an 802.2 LLC layer. We look for FFFF which isn't a used
		 * 802.2 SSAP/DSAP. This won't work for fault tolerant netware
		 * but does for the rest.
		 */
		skb->protocol = htons(ETH_P_802_3);
	else
		/*
		 * Real 802.2 LLC
		 */
		skb->protocol = htons(ETH_P_802_2);
}

/**
 * skb_vlan_tagged - check if skb is vlan tagged.
 * @skb: skbuff to query
 *
 * Returns true if the skb is tagged, regardless of whether it is hardware
 * accelerated or not.
 */
static inline bool skb_vlan_tagged(const struct sk_buff *skb)
{
	if (!skb_vlan_tag_present(skb) &&
	    likely(skb->protocol != htons(ETH_P_8021Q) &&
		   skb->protocol != htons(ETH_P_8021AD)))
		return false;

	return true;
}

/**
 * skb_vlan_tagged_multi - check if skb is vlan tagged with multiple headers.
 * @skb: skbuff to query
 *
 * Returns true if the skb is tagged with multiple vlan headers, regardless
 * of whether it is hardware accelerated or not.
 */
static inline bool skb_vlan_tagged_multi(const struct sk_buff *skb)
{
	__be16 protocol = skb->protocol;

	if (!skb_vlan_tag_present(skb)) {
		struct vlan_ethhdr *veh;

		if (likely(protocol != htons(ETH_P_8021Q) &&
			   protocol != htons(ETH_P_8021AD)))
			return false;

		veh = (struct vlan_ethhdr *)skb->data;
		protocol = veh->h_vlan_encapsulated_proto;
	}

	if (protocol != htons(ETH_P_8021Q) && protocol != htons(ETH_P_8021AD))
		return false;

	return true;
}

/**
 * vlan_features_check - drop unsafe features for skb with multiple tags.
 * @skb: skbuff to query
 * @features: features to be checked
 *
 * Returns features without unsafe ones if the skb has multiple tags.
 */
static inline netdev_features_t vlan_features_check(const struct sk_buff *skb,
						    netdev_features_t features)
{
	if (skb_vlan_tagged_multi(skb))
		features = netdev_intersect_features(features,
						     NETIF_F_SG |
						     NETIF_F_HIGHDMA |
						     NETIF_F_FRAGLIST |
						     NETIF_F_GEN_CSUM |
						     NETIF_F_HW_VLAN_CTAG_TX |
						     NETIF_F_HW_VLAN_STAG_TX);

	return features;
}

/**
 * compare_vlan_header - Compare two vlan headers
 * @h1: Pointer to vlan header
 * @h2: Pointer to vlan header
 *
 * Compare two vlan headers, returns 0 if equal.
 *
 * Please note that alignment of h1 & h2 are only guaranteed to be 16 bits.
 */
static inline unsigned long compare_vlan_header(const struct vlan_hdr *h1,
						const struct vlan_hdr *h2)
{
#if defined(CONFIG_HAVE_EFFICIENT_UNALIGNED_ACCESS)
	return *(u32 *)h1 ^ *(u32 *)h2;
#else
	return ((__force u32)h1->h_vlan_TCI ^ (__force u32)h2->h_vlan_TCI) |
	       ((__force u32)h1->h_vlan_encapsulated_proto ^
		(__force u32)h2->h_vlan_encapsulated_proto);
#endif
}
#endif /* !(_LINUX_IF_VLAN_H_) */


text:
<?php
return [
    'images' => 'Images',
    'image' => 'Image',
    'title' => 'Title',
    'null_title' => '— No title —',
    'filename' => 'Filename',
    'width' => 'Width',
    'height' => 'Height',
    'mime' => 'Mime',
    'size' => 'Size',
    'used' => 'Used',
    'user_id' => 'Author',
    'file' => 'File',
    'dimensions' => 'Dimensions',
    'column.image' => 'Image',
    'column.dimensions' => 'Dimensions',
    'column.width' => 'Width',
    'column.height' => 'Height',
    'column.mime' => 'Mime',
    'column.size' => 'Size',
    'column.used' => 'Used',
    'path' => 'File path',
    'tags' => 'Tags',
    'null_tags' => '— No Tags —',
    'library' => 'Image library',
    'library_title' => 'Images list <small>oder by updated time</small>',
    'add_new' => 'Add image',
    'upload_all' => 'Upload all',
    'replace' => 'Replace image',
    'replace_success' => 'Replace image successfully!',
    'upload_title' => 'Upload images to Server',
    'delete_title' => 'Delete image',
    'delete_success' => 'Delete image successfully!',
    'delete_confirm' => 'Are you sure you want to delete this image?',
    'delete_error' => 'Unable delete this image?',
    'upload' => 'Upload',
    'insert' => 'Insert image',
];


text:
namespace Oragon.Architecture.ApplicationHosting
{
	public enum AccountType
	{
		NetworkService,
		LocalSystem,
		LocalService,
		Custom,
		Prompt
	}
}