openairinterface5g/common/utils/threadPool/notified_fifo.h

/*
 * SPDX-License-Identifier: LicenseRef-CSSL-1.0
 */

#ifndef NOTIFIED_FIFO_H
#define NOTIFIED_FIFO_H
#include "pthread_utils.h"
#include <stdint.h>
#include <malloc.h>
#include <pthread.h>
#include "time_meas.h"
#include <memory.h>
#include <stdalign.h>
#include "assertions.h"

/// @brief Element on the notifiedFifo_t
/// next: internal FIFO chain, do not set it
/// key: a long int that the client can use to identify a job or a group of messages
/// ResponseFifo: if the client defines a response FIFO, the job will be posted back after processing
/// processingFunc: any function (of type void processingFunc(void *)) that a worker will process
/// msgData: the data passed to `processingFunc`. It can be added automatically, or you can set it to a buffer you are managing
/// malloced: a boolean that enables internal free in the case of no return FIFO or abort feature
typedef struct notifiedFIFO_elt_s {
  struct notifiedFIFO_elt_s *next;
  uint64_t key; // To filter out elements
  struct notifiedFIFO_s *reponseFifo;
  void (*processingFunc)(void *);
  bool malloced;
  oai_cputime_t creationTime;
  oai_cputime_t startProcessingTime;
  oai_cputime_t endProcessingTime;
  oai_cputime_t returnTime;
  // use alignas(32) to align msgData to 32b
  // user data behind it will be aligned to 32b as well
  // important! this needs to be the last member in the struct
  alignas(32) void *msgData;
} notifiedFIFO_elt_t;

typedef struct notifiedFIFO_s {
  notifiedFIFO_elt_t *outF;
  notifiedFIFO_elt_t *inF;
  pthread_mutex_t lockF;
  pthread_cond_t notifF;
  bool abortFIFO; // if set, the FIFO always returns NULL -> abort condition
} notifiedFIFO_t;

/// @brief Creates a new job.
/// @param size The data part of the job will have this size
/// @param key the job can be identified
/// @param reponseFifo response fifo
/// @param processingFunc function to call
/// @return new notifiedFIFO_elt_t element with extra memory allocated at the end equal to size.
static inline notifiedFIFO_elt_t *newNotifiedFIFO_elt(int size,
                                                      uint64_t key,
                                                      notifiedFIFO_t *reponseFifo,
                                                      void (*processingFunc)(void *))
{
  notifiedFIFO_elt_t *ret = (notifiedFIFO_elt_t *)memalign(32, sizeof(notifiedFIFO_elt_t) + size);
  AssertFatal(NULL != ret, "out of memory\n");
  ret->next = NULL;
  ret->key = key;
  ret->reponseFifo = reponseFifo;
  ret->processingFunc = processingFunc;
  // We set user data piece aligend 32 bytes to be able to process it with SIMD
  // msgData is aligned to 32bytes, so everything after will be as well
  ret->msgData = ((uint8_t *)ret) + sizeof(notifiedFIFO_elt_t);
  ret->malloced = true;
  return ret;
}

/// @brief Get pointer to the data carried by notifiedFIFO_elt_t
/// @param elt
/// @return void pointer to the data allocated with the message
static inline void *NotifiedFifoData(notifiedFIFO_elt_t *elt)
{
  return elt->msgData;
}

/// @brief Delete a notifiedFIFO_elt_t if its allocated
/// @param elt
static inline void delNotifiedFIFO_elt(notifiedFIFO_elt_t *elt)
{
  if (elt->malloced) {
    elt->malloced = false;
    free(elt);
  }
}

static inline void initNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf)
{
  nf->inF = NULL;
  nf->outF = NULL;
  nf->abortFIFO = false;
}
static inline void initNotifiedFIFO(notifiedFIFO_t *nf)
{
  mutexinit(nf->lockF);
  condinit(nf->notifF);
  initNotifiedFIFO_nothreadSafe(nf);
  // No delete function: the creator has only to free the memory
}

static inline void pushNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg)
{
  msg->next = NULL;

  if (nf->outF == NULL)
    nf->outF = msg;

  if (nf->inF != NULL)
    nf->inF->next = msg;

  nf->inF = msg;
}

static inline void pushNotifiedFIFO(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg)
{
  mutexlock(nf->lockF);
  if (!nf->abortFIFO) {
    pushNotifiedFIFO_nothreadSafe(nf, msg);
    condsignal(nf->notifF);
  }
  mutexunlock(nf->lockF);
}

static inline notifiedFIFO_elt_t *pullNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf)
{
  if (nf->outF == NULL)
    return NULL;
  if (nf->abortFIFO)
    return NULL;

  notifiedFIFO_elt_t *ret = nf->outF;

  AssertFatal(nf->outF != nf->outF->next, "Circular list in thread pool: push several times the same buffer is forbidden\n");

  nf->outF = nf->outF->next;

  if (nf->outF == NULL)
    nf->inF = NULL;

  return ret;
}

static inline notifiedFIFO_elt_t *pullNotifiedFIFO(notifiedFIFO_t *nf)
{
  mutexlock(nf->lockF);
  notifiedFIFO_elt_t *ret = NULL;

  while ((ret = pullNotifiedFIFO_nothreadSafe(nf)) == NULL && !nf->abortFIFO)
    condwait(nf->notifF, nf->lockF);

  mutexunlock(nf->lockF);
  return ret;
}

static inline notifiedFIFO_elt_t *pollNotifiedFIFO(notifiedFIFO_t *nf)
{
  int tmp = mutextrylock(nf->lockF);

  if (tmp != 0)
    return NULL;

  if (nf->abortFIFO) {
    mutexunlock(nf->lockF);
    return NULL;
  }

  notifiedFIFO_elt_t *ret = pullNotifiedFIFO_nothreadSafe(nf);
  mutexunlock(nf->lockF);
  return ret;
}

static inline time_stats_t exec_time_stats_NotifiedFIFO(const notifiedFIFO_elt_t *elt)
{
  time_stats_t ts = {0};
  if (elt->startProcessingTime == 0 && elt->endProcessingTime == 0)
    return ts; /* no measurements done */
  ts.in = elt->startProcessingTime;
  ts.diff = elt->endProcessingTime - ts.in;
  ts.p_time = ts.diff;
  ts.diff_square = ts.diff * ts.diff;
  ts.max = ts.diff;
  ts.trials = 1;
  return ts;
}

// This functions aborts all messages in the queue, and marks the queue as
// "aborted", such that every call to it will return NULL
static inline void abortNotifiedFIFO(notifiedFIFO_t *nf)
{
  mutexlock(nf->lockF);
  nf->abortFIFO = true;
  notifiedFIFO_elt_t **elt = &nf->outF;
  while (*elt != NULL) {
    notifiedFIFO_elt_t *p = *elt;
    *elt = (*elt)->next;
    delNotifiedFIFO_elt(p);
  }

  if (nf->outF == NULL)
    nf->inF = NULL;
  condbroadcast(nf->notifF);
  mutexunlock(nf->lockF);
}

#endif