Event Buffer Functions (bm_xxx)

Classes
struct	BUFFER_INFO
struct	EVENT_DEFRAG_BUFFER

Macros
#define	MAX_DEFRAG_EVENTS   10

Functions
static int	bm_validate_client_index_locked (bm_lock_buffer_guard &pbuf_guard)
INT	bm_match_event (short int event_id, short int trigger_mask, const EVENT_HEADER *pevent)
void	bm_remove_client_locked (BUFFER_HEADER *pheader, int j)
static void	bm_cleanup_buffer_locked (BUFFER *pbuf, const char *who, DWORD actual_time)
static void	bm_update_last_activity (DWORD millitime)
static BOOL	bm_validate_rp (const char *who, const BUFFER_HEADER *pheader, int rp)
static int	bm_incr_rp_no_check (const BUFFER_HEADER *pheader, int rp, int total_size)
static int	bm_next_rp (const char *who, const BUFFER_HEADER *pheader, const char *pdata, int rp)
static int	bm_validate_buffer_locked (const BUFFER *pbuf)
static void	bm_reset_buffer_locked (BUFFER *pbuf)
static void	bm_clear_buffer_statistics (HNDLE hDB, BUFFER *pbuf)
static void	bm_write_buffer_statistics_to_odb_copy (HNDLE hDB, const char *buffer_name, const char *client_name, int client_index, BUFFER_INFO *pbuf, BUFFER_HEADER *pheader)
static void	bm_write_buffer_statistics_to_odb (HNDLE hDB, BUFFER *pbuf, BOOL force)
INT	bm_open_buffer (const char *buffer_name, INT buffer_size, INT *buffer_handle)
INT	bm_get_buffer_handle (const char *buffer_name, INT *buffer_handle)
INT	bm_close_buffer (INT buffer_handle)
INT	bm_close_all_buffers (void)
INT	bm_write_statistics_to_odb (void)
INT	bm_set_cache_size (INT buffer_handle, size_t read_size, size_t write_size)
INT	bm_compose_event (EVENT_HEADER *event_header, short int event_id, short int trigger_mask, DWORD data_size, DWORD serial)
INT	bm_compose_event_threadsafe (EVENT_HEADER *event_header, short int event_id, short int trigger_mask, DWORD data_size, DWORD *serial)
INT	bm_add_event_request (INT buffer_handle, short int event_id, short int trigger_mask, INT sampling_type, EVENT_HANDLER *func, INT request_id)
INT	bm_request_event (HNDLE buffer_handle, short int event_id, short int trigger_mask, INT sampling_type, HNDLE *request_id, EVENT_HANDLER *func)
INT	bm_remove_event_request (INT buffer_handle, INT request_id)
INT	bm_delete_request (INT request_id)
static void	bm_validate_client_pointers_locked (const BUFFER_HEADER *pheader, BUFFER_CLIENT *pclient)
static BOOL	bm_update_read_pointer_locked (const char *caller_name, BUFFER_HEADER *pheader)
static void	bm_wakeup_producers_locked (const BUFFER_HEADER *pheader, const BUFFER_CLIENT *pc)
static void	bm_dispatch_event (int buffer_handle, EVENT_HEADER *pevent)
static void	bm_incr_read_cache_locked (BUFFER *pbuf, int total_size)
static BOOL	bm_peek_read_cache_locked (BUFFER *pbuf, EVENT_HEADER **ppevent, int *pevent_size, int *ptotal_size)
static int	bm_peek_buffer_locked (BUFFER *pbuf, BUFFER_HEADER *pheader, BUFFER_CLIENT *pc, EVENT_HEADER **ppevent, int *pevent_size, int *ptotal_size)
static void	bm_read_from_buffer_locked (const BUFFER_HEADER *pheader, int rp, char *buf, int event_size)
static void	bm_read_from_buffer_locked (const BUFFER_HEADER *pheader, int rp, std::vector< char > *vecptr, int event_size)
static BOOL	bm_check_requests (const BUFFER_CLIENT *pc, const EVENT_HEADER *pevent)
static int	bm_wait_for_more_events_locked (bm_lock_buffer_guard &pbuf_guard, BUFFER_CLIENT *pc, int timeout_msec, BOOL unlock_read_cache)
static int	bm_fill_read_cache_locked (bm_lock_buffer_guard &pbuf_guard, int timeout_msec)
static void	bm_convert_event_header (EVENT_HEADER *pevent, int convert_flags)
static int	bm_wait_for_free_space_locked (bm_lock_buffer_guard &pbuf_guard, int timeout_msec, int requested_space, bool unlock_write_cache)
static void	bm_write_to_buffer_locked (BUFFER_HEADER *pheader, int sg_n, const char *const sg_ptr[], const size_t sg_len[], size_t total_size)
static int	bm_find_first_request_locked (BUFFER_CLIENT *pc, const EVENT_HEADER *pevent)
static void	bm_notify_reader_locked (BUFFER_HEADER *pheader, BUFFER_CLIENT *pc, int old_write_pointer, int request_id)
INT	bm_send_event (INT buffer_handle, const EVENT_HEADER *pevent, int unused, int timeout_msec)
int	bm_send_event_vec (int buffer_handle, const std::vector< char > &event, int timeout_msec)
int	bm_send_event_vec (int buffer_handle, const std::vector< std::vector< char > > &event, int timeout_msec)
static INT	bm_flush_cache_locked (bm_lock_buffer_guard &pbuf_guard, int timeout_msec)
int	bm_send_event_sg (int buffer_handle, int sg_n, const char *const sg_ptr[], const size_t sg_len[], int timeout_msec)
static int	bm_flush_cache_rpc (int buffer_handle, int timeout_msec)
INT	bm_flush_cache (int buffer_handle, int timeout_msec)
static INT	bm_read_buffer (BUFFER *pbuf, INT buffer_handle, void **bufptr, void *buf, INT *buf_size, std::vector< char > *vecptr, int timeout_msec, int convert_flags, BOOL dispatch)
static INT	bm_receive_event_rpc (INT buffer_handle, void *buf, int *buf_size, EVENT_HEADER **ppevent, std::vector< char > *pvec, int timeout_msec)
INT	bm_receive_event (INT buffer_handle, void *destination, INT *buf_size, int timeout_msec)
INT	bm_receive_event_alloc (INT buffer_handle, EVENT_HEADER **ppevent, int timeout_msec)
INT	bm_receive_event_vec (INT buffer_handle, std::vector< char > *pvec, int timeout_msec)
static int	bm_skip_event (BUFFER *pbuf)
INT	bm_skip_event (INT buffer_handle)
static INT	bm_push_buffer (BUFFER *pbuf, int buffer_handle)
INT	bm_check_buffers ()
INT	bm_poll_event ()
INT	bm_empty_buffers ()

Variables
static DWORD	_bm_max_event_size = 0
static int	_bm_lock_timeout = 5 * 60 * 1000
static double	_bm_mutex_timeout_sec = _bm_lock_timeout/1000 + 15.000
static EVENT_DEFRAG_BUFFER	defrag_buffer [MAX_DEFRAG_EVENTS]

Detailed Description

dox dox

Macro Definition Documentation

MAX_DEFRAG_EVENTS

#define MAX_DEFRAG_EVENTS   10

dox

Definition at line 11280 of file midas.cxx.

Function Documentation

bm_add_event_request()

INT bm_add_event_request	(	INT	buffer_handle,
		short int	event_id,
		short int	trigger_mask,
		INT	sampling_type,
		EVENT_HANDLER *	func,
		INT	request_id
	)

dox

Definition at line 8322 of file midas.cxx.

{
   if (rpc_is_remote())
      return rpc_call(RPC_BM_ADD_EVENT_REQUEST, buffer_handle, event_id,
                      trigger_mask, sampling_type, (INT) (POINTER_T) func, request_id);
 
#ifdef LOCAL_ROUTINES
   {
      int status = 0;
 
      BUFFER *pbuf = bm_get_buffer("bm_add_event_request", buffer_handle, &status);
 
      if (!pbuf)
         return status;
 
      /* lock buffer */
      bm_lock_buffer_guard pbuf_guard(pbuf);
 
      if (!pbuf_guard.is_locked())
         return pbuf_guard.get_status();
 
      /* avoid callback/non callback requests */
      if (func == NULL && pbuf->callback) {
         pbuf_guard.unlock(); // unlock before cm_msg()
         cm_msg(MERROR, "bm_add_event_request", "mixing callback/non callback requests not possible");
         return BM_INVALID_MIXING;
      }
 
      /* do not allow GET_RECENT with nonzero cache size */
      if (sampling_type == GET_RECENT && pbuf->read_cache_size > 0) {
         pbuf_guard.unlock(); // unlock before cm_msg()
         cm_msg(MERROR, "bm_add_event_request", "GET_RECENT request not possible if read cache is enabled");
         return BM_INVALID_PARAM;
      }
 
      /* get a pointer to the proper client structure */
      BUFFER_CLIENT *pclient = bm_get_my_client_locked(pbuf_guard);
 
      /* look for a empty request entry */
      int i;
      for (i = 0; i < MAX_EVENT_REQUESTS; i++)
         if (!pclient->event_request[i].valid)
            break;
 
      if (i == MAX_EVENT_REQUESTS) {
         // implicit unlock
         return BM_NO_MEMORY;
      }
 
      /* setup event_request structure */
      pclient->event_request[i].id = request_id;
      pclient->event_request[i].valid = TRUE;
      pclient->event_request[i].event_id = event_id;
      pclient->event_request[i].trigger_mask = trigger_mask;
      pclient->event_request[i].sampling_type = sampling_type;
 
      pclient->all_flag = pclient->all_flag || (sampling_type & GET_ALL);
 
      pbuf->get_all_flag = pclient->all_flag;
 
      /* set callback flag in buffer structure */
      if (func != NULL)
         pbuf->callback = TRUE;
 
      /*
         Save the index of the last request in the list so that later only the
         requests 0..max_request_index-1 have to be searched through.
       */
 
      if (i + 1 > pclient->max_request_index)
         pclient->max_request_index = i + 1;
   }
#endif                          /* LOCAL_ROUTINES */
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_check_buffers()

INT bm_check_buffers

(

void

)

Check if any requested event is waiting in a buffer

Returns

TRUE More events are waiting
FALSE No more events are waiting

Definition at line 10962 of file midas.cxx.

                       {
#ifdef LOCAL_ROUTINES
   {
      INT status = 0;
      BOOL bMore;
      DWORD start_time;
      //static DWORD last_time = 0;
 
      /* if running as a server, buffer checking is done by client
         via ASYNC bm_receive_event */
      if (rpc_is_mserver()) {
         return FALSE;
      }
 
      bMore = FALSE;
      start_time = ss_millitime();
 
      std::vector<BUFFER*> mybuffers;
      
      gBuffersMutex.lock();
      mybuffers = gBuffers;
      gBuffersMutex.unlock();
 
      /* go through all buffers */
      for (size_t idx = 0; idx < mybuffers.size(); idx++) {
         BUFFER* pbuf = mybuffers[idx];
 
         if (!pbuf || !pbuf->attached)
            continue;
 
         //int count_loops = 0;
         while (1) {
            if (pbuf->attached) {
               /* one bm_push_event could cause a run stop and a buffer close, which
                * would crash the next call to bm_push_event(). So check for valid
                * buffer on each call */
 
               /* this is what happens:
                * bm_push_buffer() may call a user callback function
                * user callback function may indirectly call bm_close() of this buffer,
                * i.e. if it stops the run,
                * bm_close() will set pbuf->attached to false, but will not delete pbuf or touch gBuffers
                * here we will see pbuf->attched is false and quit this loop
                */
 
               status = bm_push_buffer(pbuf, idx + 1);
 
               if (status == BM_CORRUPTED) {
                  return status;
               }
 
               //printf("bm_check_buffers: bm_push_buffer() returned %d, loop %d, time %d\n", status, count_loops, ss_millitime() - start_time);
 
               if (status != BM_MORE_EVENTS) {
                  //DWORD t = ss_millitime() - start_time;
                  //printf("bm_check_buffers: index %d, period %d, elapsed %d, loop %d, no more events\n", idx, start_time - last_time, t, count_loops);
                  break;
               }
 
               // count_loops++;
            }
 
            // NB: this code has a logic error: if 2 buffers always have data,
            // this timeout will cause us to exit reading the 1st buffer
            // after 1000 msec, then we read the 2nd buffer exactly once,
            // and exit the loop because the timeout is still active -
            // we did not reset "start_time" when we started reading
            // from the 2nd buffer. Result is that we always read all
            // the data in a loop from the 1st buffer, but read just
            // one event from the 2nd buffer, resulting in severe unfairness.
 
            /* stop after one second */
            DWORD t = ss_millitime() - start_time;
            if (t > 1000) {
               //printf("bm_check_buffers: index %d, period %d, elapsed %d, loop %d, timeout.\n", idx, start_time - last_time, t, count_loops);
               bMore = TRUE;
               break;
            }
         }
      }
 
      //last_time = start_time;
 
      return bMore;
 
   }
#else                           /* LOCAL_ROUTINES */
 
   return FALSE;
 
#endif
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_check_requests()

static BOOL bm_check_requests ( const BUFFER_CLIENT *  pc, const EVENT_HEADER *  pevent  )

static

Definition at line 8974 of file midas.cxx.

                                                                                   {
 
   BOOL is_requested = FALSE;
   int i;
   for (i = 0; i < pc->max_request_index; i++) {
      const EVENT_REQUEST *prequest = pc->event_request + i;
      if (prequest->valid) {
         if (bm_match_event(prequest->event_id, prequest->trigger_mask, pevent)) {
            /* check if this is a recent event */
            if (prequest->sampling_type == GET_RECENT) {
               if (ss_time() - pevent->time_stamp > 1) {
                  /* skip that event */
                  continue;
               }
            }
 
            is_requested = TRUE;
            break;
         }
      }
   }
   return is_requested;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_cleanup_buffer_locked()

static void bm_cleanup_buffer_locked ( BUFFER *  pbuf, const char *  who, DWORD  actual_time  )

static

Check all clients on buffer, remove invalid clients

Definition at line 6074 of file midas.cxx.

                                                                                       {
   BUFFER_HEADER *pheader;
   BUFFER_CLIENT *pbclient;
   int j;
 
   pheader = pbuf->buffer_header;
   pbclient = pheader->client;
 
   /* now check other clients */
   for (j = 0; j < pheader->max_client_index; j++, pbclient++) {
      if (pbclient->pid) {
         if (!ss_pid_exists(pbclient->pid)) {
            cm_msg(MINFO, "bm_cleanup",
                   "Client \'%s\' on buffer \'%s\' removed by %s because process pid %d does not exist", pbclient->name,
                   pheader->name, who, pbclient->pid);
 
            bm_remove_client_locked(pheader, j);
            continue;
         }
      }
 
      /* If client process has no activity, clear its buffer entry. */
      if (pbclient->pid && pbclient->watchdog_timeout > 0) {
         DWORD tdiff = actual_time - pbclient->last_activity;
#if 0
         printf("buffer [%s] client [%-32s] times 0x%08x 0x%08x, diff 0x%08x %5d, timeout %d\n",
                pheader->name,
                pbclient->name,
                pbclient->last_activity,
                actual_time,
                tdiff,
                tdiff,
                pbclient->watchdog_timeout);
#endif
         if (actual_time > pbclient->last_activity &&
             tdiff > pbclient->watchdog_timeout) {
 
            cm_msg(MINFO, "bm_cleanup", "Client \'%s\' on buffer \'%s\' removed by %s (idle %1.1lfs, timeout %1.0lfs)",
                   pbclient->name, pheader->name, who,
                   tdiff / 1000.0,
                   pbclient->watchdog_timeout / 1000.0);
 
            bm_remove_client_locked(pheader, j);
         }
      }
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_clear_buffer_statistics()

static void bm_clear_buffer_statistics ( HNDLE  hDB, BUFFER *  pbuf  )

static

Definition at line 6416 of file midas.cxx.

                                                                {
   HNDLE hKey;
   int status;
 
   char str[256 + 2 * NAME_LENGTH];
   sprintf(str, "/System/buffers/%s/Clients/%s/writes_blocked_by", pbuf->buffer_name, pbuf->client_name);
   //printf("delete [%s]\n", str);
   status = db_find_key(hDB, 0, str, &hKey);
   if (status == DB_SUCCESS) {
      status = db_delete_key(hDB, hKey, FALSE);
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_close_all_buffers()

INT bm_close_all_buffers

(

void

)

Close all open buffers

Returns

BM_SUCCESS

Definition at line 7251 of file midas.cxx.

                               {
   if (rpc_is_remote())
      return rpc_call(RPC_BM_CLOSE_ALL_BUFFERS);
 
#ifdef LOCAL_ROUTINES
   {
      cm_msg_close_buffer();
 
      gBuffersMutex.lock();
      size_t nbuf = gBuffers.size();
      gBuffersMutex.unlock();
 
      for (size_t i = nbuf; i > 0; i--) {
         bm_close_buffer(i);
      }
 
      gBuffersMutex.lock();
      for (size_t i=0; i< gBuffers.size(); i++) {
         BUFFER* pbuf = gBuffers[i];
         if (!pbuf)
            continue;
         delete pbuf;
         pbuf = NULL;
         gBuffers[i] = NULL;
      }
      gBuffersMutex.unlock();
   }
#endif                          /* LOCAL_ROUTINES */
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_close_buffer()

INT bm_close_buffer

(

INT

buffer_handle

)

Closes an event buffer previously opened with bm_open_buffer().

Parameters

buffer_handle

buffer handle

Returns

BM_SUCCESS, BM_INVALID_HANDLE

Definition at line 7104 of file midas.cxx.

                                       {
   //printf("bm_close_buffer: handle %d\n", buffer_handle);
 
   if (rpc_is_remote())
      return rpc_call(RPC_BM_CLOSE_BUFFER, buffer_handle);
 
#ifdef LOCAL_ROUTINES
   {
      int status = 0;
 
      BUFFER *pbuf = bm_get_buffer(NULL, buffer_handle, &status);
 
      if (!pbuf)
         return status;
 
      //printf("bm_close_buffer: handle %d, name [%s]\n", buffer_handle, pheader->name);
 
      int i;
 
      { /* delete all requests for this buffer */
         _request_list_mutex.lock();
         std::vector<EventRequest> request_list_copy = _request_list;
         _request_list_mutex.unlock();
         for (size_t i = 0; i < request_list_copy.size(); i++) {
            if (request_list_copy[i].buffer_handle == buffer_handle) {
               bm_delete_request(i);
            }
         }
      }
 
      HNDLE hDB;
      cm_get_experiment_database(&hDB, NULL);
 
      if (hDB) {
         /* write statistics to odb */
         bm_write_buffer_statistics_to_odb(hDB, pbuf, TRUE);
      }
 
      /* lock buffer in correct order */
 
      status = bm_lock_buffer_read_cache(pbuf);
 
      if (status != BM_SUCCESS) {
         return status;
      }
 
      status = bm_lock_buffer_write_cache(pbuf);
 
      if (status != BM_SUCCESS) {
         pbuf->read_cache_mutex.unlock();
         return status;
      }
 
      bm_lock_buffer_guard pbuf_guard(pbuf);
 
      if (!pbuf_guard.is_locked()) {
         pbuf->write_cache_mutex.unlock();
         pbuf->read_cache_mutex.unlock();
         return pbuf_guard.get_status();
      }
 
      BUFFER_HEADER *pheader = pbuf->buffer_header;
 
      /* mark entry in _buffer as empty */
      pbuf->attached = false;
 
      BUFFER_CLIENT* pclient = bm_get_my_client_locked(pbuf_guard);
 
      if (pclient) {
         /* clear entry from client structure in buffer header */
         memset(pclient, 0, sizeof(BUFFER_CLIENT));
      }
 
      /* calculate new max_client_index entry */
      for (i = MAX_CLIENTS - 1; i >= 0; i--)
         if (pheader->client[i].pid != 0)
            break;
      pheader->max_client_index = i + 1;
 
      /* count new number of clients */
      int j = 0;
      for (i = MAX_CLIENTS - 1; i >= 0; i--)
         if (pheader->client[i].pid != 0)
            j++;
      pheader->num_clients = j;
 
      int destroy_flag = (pheader->num_clients == 0);
 
      // we hold the locks on the read cache and the write cache.
 
      /* free cache */
      if (pbuf->read_cache_size > 0) {
         free(pbuf->read_cache);
         pbuf->read_cache = NULL;
         pbuf->read_cache_size = 0;
         pbuf->read_cache_rp = 0;
         pbuf->read_cache_wp = 0;
      }
 
      if (pbuf->write_cache_size > 0) {
         free(pbuf->write_cache);
         pbuf->write_cache = NULL;
         pbuf->write_cache_size = 0;
         pbuf->write_cache_rp = 0;
         pbuf->write_cache_wp = 0;
      }
 
      /* check if anyone is waiting and wake him up */
 
      for (int i = 0; i < pheader->max_client_index; i++) {
         BUFFER_CLIENT *pclient = pheader->client + i;
         if (pclient->pid && (pclient->write_wait || pclient->read_wait))
            ss_resume(pclient->port, "B  ");
      }
 
      /* unmap shared memory, delete it if we are the last */
 
      ss_shm_close(pbuf->buffer_name, pbuf->buffer_header, pbuf->shm_size, pbuf->shm_handle, destroy_flag);
 
      /* after ss_shm_close() these are invalid: */
 
      pheader = NULL;
      pbuf->buffer_header = NULL;
      pbuf->shm_size = 0;
      pbuf->shm_handle = 0;
 
      /* unlock buffer in correct order */
 
      pbuf_guard.unlock();
 
      pbuf->write_cache_mutex.unlock();
      pbuf->read_cache_mutex.unlock();
 
      /* delete semaphore */
 
      ss_semaphore_delete(pbuf->semaphore, destroy_flag);
   }
#endif                          /* LOCAL_ROUTINES */
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_compose_event()

INT bm_compose_event	(	EVENT_HEADER *	event_header,
		short int	event_id,
		short int	trigger_mask,
		DWORD	data_size,
		DWORD	serial
	)

Compose a Midas event header. An event header can usually be set-up manually or through this routine. If the data size of the event is not known when the header is composed, it can be set later with event_header->data-size = <...> Following structure is created at the beginning of an event

typedef struct {
 short int     event_id;
 short int     trigger_mask;
 DWORD         serial_number;
 DWORD         time_stamp;
 DWORD         data_size;
} EVENT_HEADER;
 
char event[1000];
 bm_compose_event((EVENT_HEADER *)event, 1, 0, 100, 1);
 *(event+sizeof(EVENT_HEADER)) = <...>

Parameters

event_header	pointer to the event header
event_id	event ID of the event
trigger_mask	trigger mask of the event
data_size	size if the data part of the event in bytes
serial	serial number

Returns

BM_SUCCESS

Definition at line 8289 of file midas.cxx.

{
   event_header->event_id = event_id;
   event_header->trigger_mask = trigger_mask;
   event_header->data_size = data_size;
   event_header->time_stamp = ss_time();
   event_header->serial_number = serial;
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_compose_event_threadsafe()

INT bm_compose_event_threadsafe	(	EVENT_HEADER *	event_header,
		short int	event_id,
		short int	trigger_mask,
		DWORD	data_size,
		DWORD *	serial
	)

Definition at line 8300 of file midas.cxx.

{
   static std::mutex mutex;
 
   event_header->event_id = event_id;
   event_header->trigger_mask = trigger_mask;
   event_header->data_size = data_size;
   event_header->time_stamp = ss_time();
   {
      std::lock_guard<std::mutex> lock(mutex);
      event_header->serial_number = *serial;
      *serial = *serial + 1;
      // implicit unlock
   }
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_convert_event_header()

static void bm_convert_event_header ( EVENT_HEADER *  pevent, int  convert_flags  )

static

Definition at line 9077 of file midas.cxx.

                                                                             {
   /* now convert event header */
   if (convert_flags) {
      rpc_convert_single(&pevent->event_id, TID_INT16, RPC_OUTGOING, convert_flags);
      rpc_convert_single(&pevent->trigger_mask, TID_INT16, RPC_OUTGOING, convert_flags);
      rpc_convert_single(&pevent->serial_number, TID_UINT32, RPC_OUTGOING, convert_flags);
      rpc_convert_single(&pevent->time_stamp, TID_UINT32, RPC_OUTGOING, convert_flags);
      rpc_convert_single(&pevent->data_size, TID_UINT32, RPC_OUTGOING, convert_flags);
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_delete_request()

INT bm_delete_request

(

INT

request_id

)

Deletes an event request previously done with bm_request_event(). When an event request gets deleted, events of that requested type are not received any more. When a buffer is closed via bm_close_buffer(), all event requests from that buffer are deleted automatically

Parameters

request_id

request identifier given by bm_request_event()

Returns

BM_SUCCESS, BM_INVALID_HANDLE

Definition at line 8592 of file midas.cxx.

{
   _request_list_mutex.lock();
   
   if (request_id < 0 || size_t(request_id) >= _request_list.size()) {
      _request_list_mutex.unlock();
      return BM_INVALID_HANDLE;
   }
 
   int buffer_handle = _request_list[request_id].buffer_handle;
 
   _request_list[request_id].clear();
 
   _request_list_mutex.unlock();
 
   /* remove request entry from buffer */
   return bm_remove_event_request(buffer_handle, request_id);
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_dispatch_event()

static void bm_dispatch_event ( int  buffer_handle, EVENT_HEADER *  pevent  )

static

Definition at line 8831 of file midas.cxx.

{
   _request_list_mutex.lock();
   bool locked = true;
   size_t n = _request_list.size();
   /* call dispatcher */
   for (size_t i = 0; i < n; i++) {
      if (!locked) {
         _request_list_mutex.lock();
         locked = true;
      }
      EventRequest r = _request_list[i];
      if (r.buffer_handle != buffer_handle)
         continue;
      if (!bm_match_event(r.event_id, r.trigger_mask, pevent))
         continue;
      /* must release the lock on the request list: user provided r.dispatcher() can add or remove event requests, and we will deadlock. K.O. */
      _request_list_mutex.unlock();
      locked = false;
      /* if event is fragmented, call defragmenter */
      if (((uint16_t(pevent->event_id) & uint16_t(0xF000)) == uint16_t(EVENTID_FRAG1)) || ((uint16_t(pevent->event_id) & uint16_t(0xF000)) == uint16_t(EVENTID_FRAG))) {
         bm_defragment_event(buffer_handle, i, pevent, (void *) (pevent + 1), r.dispatcher);
      } else {
         r.dispatcher(buffer_handle, i, pevent, (void *) (pevent + 1));
      }
   }
   if (locked)
      _request_list_mutex.unlock();
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_empty_buffers()

INT bm_empty_buffers

(

void

)

Clears event buffer and cache. If an event buffer is large and a consumer is slow in analyzing events, events are usually received some time after they are produced. This effect is even more experienced if a read cache is used (via bm_set_cache_size()). When changes to the hardware are made in the experience, the consumer will then still analyze old events before any new event which reflects the hardware change. Users can be fooled by looking at histograms which reflect the hardware change many seconds after they have been made.

To overcome this potential problem, the analyzer can call bm_empty_buffers() just after the hardware change has been made which skips all old events contained in event buffers and read caches. Technically this is done by forwarding the read pointer of the client. No events are really deleted, they are still visible to other clients like the logger.

Note that the front-end also contains write buffers which can delay the delivery of events. The standard front-end framework mfe.c reduces this effect by flushing all buffers once every second.

Returns

BM_SUCCESS

Definition at line 11248 of file midas.cxx.

                       {
   if (rpc_is_remote())
      return rpc_call(RPC_BM_EMPTY_BUFFERS);
 
#ifdef LOCAL_ROUTINES
   {
      std::vector<BUFFER*> mybuffers;
      
      gBuffersMutex.lock();
      mybuffers = gBuffers;
      gBuffersMutex.unlock();
 
      /* go through all buffers */
      for (BUFFER* pbuf : mybuffers) {
         if (!pbuf)
            continue;
         if (!pbuf->attached)
            continue;
 
         int status = bm_skip_event(pbuf);
         if (status != BM_SUCCESS)
            return status;
      }
   }
#endif                          /* LOCAL_ROUTINES */
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_fill_read_cache_locked()

static int bm_fill_read_cache_locked ( bm_lock_buffer_guard &  pbuf_guard, int  timeout_msec  )

static

Definition at line 9000 of file midas.cxx.

{
   BUFFER* pbuf = pbuf_guard.get_pbuf();
   BUFFER_HEADER* pheader = pbuf->buffer_header;
   BUFFER_CLIENT *pc = bm_get_my_client_locked(pbuf_guard);
   BOOL need_wakeup = FALSE;
 
   //printf("bm_fill_read_cache: [%s] timeout %d, size %d, rp %d, wp %d\n", pheader->name, timeout_msec, pbuf->read_cache_size, pbuf->read_cache_rp, pbuf->read_cache_wp);
 
   /* loop over all events in the buffer */
 
   while (1) {
      EVENT_HEADER *pevent = NULL;
      int event_size = 3; // poison value
      int total_size = 3; // poison value
 
      int status = bm_peek_buffer_locked(pbuf, pheader, pc, &pevent, &event_size, &total_size);
      if (status == BM_CORRUPTED) {
         return status;
      } else if (status != BM_SUCCESS) {
         /* event buffer is empty */
         if (timeout_msec == BM_NO_WAIT) {
            if (need_wakeup)
               bm_wakeup_producers_locked(pheader, pc);
            if (pbuf->read_cache_rp == pbuf->read_cache_wp) {
               // read cache is empty
               return BM_ASYNC_RETURN;
            }
            return BM_SUCCESS;
         }
 
         int status = bm_wait_for_more_events_locked(pbuf_guard, pc, timeout_msec, TRUE);
 
         if (status != BM_SUCCESS) {
            // we only come here with SS_ABORT & co
            return status;
         }
 
         // make sure we wait for new event only once
         timeout_msec = BM_NO_WAIT;
         // go back to bm_peek_buffer_locked
         continue;
      }
 
      /* loop over all requests: if this event matches a request,
       * copy it to the read cache */
 
      BOOL is_requested = bm_check_requests(pc, pevent);
 
      if (is_requested) {
         if (pbuf->read_cache_wp + total_size > pbuf->read_cache_size) {
            /* read cache is full */
            if (need_wakeup)
               bm_wakeup_producers_locked(pheader, pc);
            return BM_SUCCESS;
         }
 
         bm_read_from_buffer_locked(pheader, pc->read_pointer, pbuf->read_cache + pbuf->read_cache_wp, event_size);
 
         pbuf->read_cache_wp += total_size;
 
         /* update statistics */
         pheader->num_out_events++;
         pbuf->count_read++;
         pbuf->bytes_read += event_size;
      }
 
      /* shift read pointer */
 
      int new_read_pointer = bm_incr_rp_no_check(pheader, pc->read_pointer, total_size);
      pc->read_pointer = new_read_pointer;
 
      need_wakeup = TRUE;
   }
   /* NOT REACHED */
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_find_first_request_locked()

static int bm_find_first_request_locked ( BUFFER_CLIENT *  pc, const EVENT_HEADER *  pevent  )

static

Definition at line 9597 of file midas.cxx.

                                                                                       {
   if (pc->pid) {
      int j;
      for (j = 0; j < pc->max_request_index; j++) {
         const EVENT_REQUEST *prequest = pc->event_request + j;
         if (prequest->valid && bm_match_event(prequest->event_id, prequest->trigger_mask, pevent)) {
            return prequest->id;
         }
      }
   }
 
   return -1;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_flush_cache()

INT bm_flush_cache	(	int	buffer_handle,
		int	timeout_msec
	)

Definition at line 10215 of file midas.cxx.

{
   if (rpc_is_remote()) {
      return bm_flush_cache_rpc(buffer_handle, timeout_msec);
   }
 
#ifdef LOCAL_ROUTINES
   {
      INT status = 0;
 
      //printf("bm_flush_cache!\n");
 
      BUFFER *pbuf = bm_get_buffer("bm_flush_cache", buffer_handle, &status);
 
      if (!pbuf)
         return status;
 
      if (pbuf->write_cache_size == 0)
         return BM_SUCCESS;
 
      status = bm_lock_buffer_write_cache(pbuf);
 
      if (status != BM_SUCCESS)
         return status;
 
      /* check if anything needs to be flushed */
      if (pbuf->write_cache_wp == 0) {
         pbuf->write_cache_mutex.unlock();
         return BM_SUCCESS;
      }
 
      /* lock the buffer */
      bm_lock_buffer_guard pbuf_guard(pbuf);
 
      if (!pbuf_guard.is_locked())
         return pbuf_guard.get_status();
 
      status = bm_flush_cache_locked(pbuf_guard, timeout_msec);
 
      /* unlock in correct order */
 
      if (pbuf_guard.is_locked()) {
         // check if bm_wait_for_free_space() failed to relock the buffer
         pbuf_guard.unlock();
      }
 
      pbuf->write_cache_mutex.unlock();
 
      return status;
   }
#endif                          /* LOCAL_ROUTINES */
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_flush_cache_locked()

static INT bm_flush_cache_locked ( bm_lock_buffer_guard &  pbuf_guard, int  timeout_msec  )

static

Empty write cache. This function should be used if events in the write cache should be visible to the consumers immediately. It should be called at the end of each run, otherwise events could be kept in the write buffer and will flow to the data of the next run.

Parameters

buffer_handle	Buffer handle obtained via bm_open_buffer() or 0 to flush data in the mserver event socket
timeout_msec	Timeout waiting for free space in the event buffer. If BM_WAIT, wait forever. If BM_NO_WAIT, the function returns immediately with a value of BM_ASYNC_RETURN without writing the cache.

Returns

BM_SUCCESS, BM_INVALID_HANDLE
BM_ASYNC_RETURN Routine called with async_flag == BM_NO_WAIT and buffer has not enough space to receive cache
BM_NO_MEMORY Event is too large for network buffer or event buffer. One has to increase the event buffer size "/Experiment/Buffer sizes/SYSTEM" and/or /Experiment/MAX_EVENT_SIZE in ODB.

Definition at line 10071 of file midas.cxx.

{
   // NB we come here with write cache locked and buffer locked.
 
   {
      INT status = 0;
 
      //printf("bm_flush_cache_locked!\n");
 
      BUFFER* pbuf = pbuf_guard.get_pbuf();
      BUFFER_HEADER* pheader = pbuf->buffer_header;
 
      //printf("bm_flush_cache_locked: buffer %s, cache rp %zu, wp %zu, timeout %d msec\n", pbuf->buffer_name, pbuf->write_cache_rp, pbuf->write_cache_wp, timeout_msec);
 
      int old_write_pointer = pheader->write_pointer;
 
      int request_id[MAX_CLIENTS];
      for (int i = 0; i < pheader->max_client_index; i++) {
         request_id[i] = -1;
      }
         
      size_t ask_rp = pbuf->write_cache_rp;
      size_t ask_wp = pbuf->write_cache_wp;
 
      if (ask_wp == 0) { // nothing to do
         return BM_SUCCESS;
      }
 
      if (ask_rp == ask_wp) { // nothing to do
         return BM_SUCCESS;
      }
 
      assert(ask_rp < ask_wp);
 
      size_t ask_free = ALIGN8(ask_wp - ask_rp);
 
      if (ask_free == 0) { // nothing to do
         return BM_SUCCESS;
      }
 
#if 0
      status = bm_validate_buffer_locked(pbuf);
      if (status != BM_SUCCESS) {
         printf("bm_flush_cache: corrupted 111!\n");
         abort();
      }
#endif
 
      status = bm_wait_for_free_space_locked(pbuf_guard, timeout_msec, ask_free, true);
 
      if (status != BM_SUCCESS) {
         return status;
      }
 
      // NB: ask_rp, ask_wp and ask_free are invalid after calling bm_wait_for_free_space():
      //
      // wait_for_free_space() will sleep with all locks released,
      // during this time, another thread may call bm_send_event() that will
      // add one or more events to the write cache and after wait_for_free_space()
      // returns, size of data in cache will be bigger than the amount
      // of free space we requested. so we need to keep track of how
      // much data we write to the buffer and ask for more data
      // if we run short. This is the reason for the big loop
      // around wait_for_free_space(). We ask for slightly too little free
      // space to make sure all this code is always used and does work. K.O.
 
      if (pbuf->write_cache_wp == 0) {
         /* somebody emptied the cache while we were inside bm_wait_for_free_space */
         return BM_SUCCESS;
      }
 
      //size_t written = 0;
      while (pbuf->write_cache_rp < pbuf->write_cache_wp) {
         /* loop over all events in cache */
 
         const EVENT_HEADER *pevent = (const EVENT_HEADER *) (pbuf->write_cache + pbuf->write_cache_rp);
         size_t event_size = (pevent->data_size + sizeof(EVENT_HEADER));
         size_t total_size = ALIGN8(event_size);
 
#if 0
         printf("bm_flush_cache: cache size %d, wp %d, rp %d, event data_size %d, event_size %d, total_size %d, free %d, written %d\n",
                int(pbuf->write_cache_size),
                int(pbuf->write_cache_wp),
                int(pbuf->write_cache_rp),
                int(pevent->data_size),
                int(event_size),
                int(total_size),
                int(ask_free),
                int(written));
#endif
 
         // check for crazy event size
         assert(total_size >= sizeof(EVENT_HEADER));
         assert(total_size <= (size_t)pheader->size);
 
         bm_write_to_buffer_locked(pheader, 1, (char**)&pevent, &event_size, total_size);
 
         /* update statistics */
         pheader->num_in_events++;
         pbuf->count_sent += 1;
         pbuf->bytes_sent += total_size;
 
         /* see comment for the same code in bm_send_event().
          * We make sure the buffer is never 100% full */
         assert(pheader->write_pointer != pheader->read_pointer);
 
         /* check if anybody has a request for this event */
         for (int i = 0; i < pheader->max_client_index; i++) {
            BUFFER_CLIENT *pc = pheader->client + i;
            int r = bm_find_first_request_locked(pc, pevent);
            if (r >= 0) {
               request_id[i] = r;
            }
         }
            
         /* this loop does not loop forever because rp
          * is monotonously incremented here. write_cache_wp does
          * not change */
 
         pbuf->write_cache_rp += total_size;
         //written += total_size;
 
         assert(pbuf->write_cache_rp > 0);
         assert(pbuf->write_cache_rp <= pbuf->write_cache_size);
         assert(pbuf->write_cache_rp <= pbuf->write_cache_wp);
      }
 
      /* the write cache is now empty */
      assert(pbuf->write_cache_wp == pbuf->write_cache_rp);
      pbuf->write_cache_wp = 0;
      pbuf->write_cache_rp = 0;
 
      /* check which clients are waiting */
      for (int i = 0; i < pheader->max_client_index; i++) {
         BUFFER_CLIENT *pc = pheader->client + i;
         bm_notify_reader_locked(pheader, pc, old_write_pointer, request_id[i]);
      }
   }
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_flush_cache_rpc()

static int bm_flush_cache_rpc ( int  buffer_handle, int  timeout_msec  )

static

Definition at line 9995 of file midas.cxx.

{
   //printf("bm_flush_cache_rpc: handle %d, timeout %d\n", buffer_handle, timeout_msec);
 
   DWORD time_start = ss_millitime();
   DWORD time_end = time_start + timeout_msec;
   
   int xtimeout_msec = timeout_msec;
 
   while (1) {
      if (timeout_msec == BM_WAIT) {
         xtimeout_msec = 1000;
      } else if (timeout_msec == BM_NO_WAIT) {
         xtimeout_msec = BM_NO_WAIT;
      } else {
         if (xtimeout_msec > 1000) {
            xtimeout_msec = 1000;
         }
      }
 
      int status = rpc_call(RPC_BM_FLUSH_CACHE, buffer_handle, xtimeout_msec);
      
      //printf("bm_flush_cache_rpc: handle %d, timeout %d, status %d\n", buffer_handle, xtimeout_msec, status);
 
      if (status == BM_ASYNC_RETURN) {
         if (timeout_msec == BM_WAIT) {
            // BM_WAIT means wait forever
            continue;
         } else if (timeout_msec == BM_NO_WAIT) {
            // BM_NO_WAIT means do not wait
            return status;
         } else {
            DWORD now = ss_millitime();
            if (now >= time_end) {
               // timeout, return BM_ASYNC_RETURN
               return status;
            }
 
            DWORD remain = time_end - now;
 
            if (remain < (DWORD)xtimeout_msec) {
               xtimeout_msec = remain;
            }
 
            // keep asking for event...
            continue;
         }
      } else if (status == BM_SUCCESS) {
         // success, return BM_SUCCESS
         return status;
      } else {
         // error
         return status;
      }
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_get_buffer_handle()

INT bm_get_buffer_handle	(	const char *	buffer_name,
		INT *	buffer_handle
	)

If buffer is already open, return it's handle

Parameters

buffer_name

buffer name

Returns

BM_SUCCESS, BM_NOT_FOUND

Definition at line 7083 of file midas.cxx.

{
   gBuffersMutex.lock();
   for (size_t i = 0; i < gBuffers.size(); i++) {
      BUFFER* pbuf = gBuffers[i];
      if (pbuf && pbuf->attached && equal_ustring(pbuf->buffer_name, buffer_name)) {
         *buffer_handle = i + 1;
         gBuffersMutex.unlock();
         return BM_SUCCESS;
      }
   }
   gBuffersMutex.unlock();
   return BM_NOT_FOUND;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_incr_read_cache_locked()

static void bm_incr_read_cache_locked ( BUFFER *  pbuf, int  total_size  )

static

Definition at line 8863 of file midas.cxx.

                                                                    {
   /* increment read cache read pointer */
   pbuf->read_cache_rp += total_size;
 
   if (pbuf->read_cache_rp == pbuf->read_cache_wp) {
      pbuf->read_cache_rp = 0;
      pbuf->read_cache_wp = 0;
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_incr_rp_no_check()

static int bm_incr_rp_no_check ( const BUFFER_HEADER *  pheader, int  rp, int  total_size  )

static

Definition at line 6231 of file midas.cxx.

{
#if 0
   if (gRpLog == NULL) {
      gRpLog = fopen("rp.log", "a");
   }
   if (gRpLog && (total_size < 16)) {
      const char *pdata = (const char *) (pheader + 1);
      const DWORD *pevent = (const DWORD*) (pdata + rp);
      fprintf(gRpLog, "%s: rp %d, total_size %d, at rp 0x%08x 0x%08x 0x%08x 0x%08x 0x%08x 0x%08x\n", pheader->name, rp, total_size,
              pevent[0], pevent[1], pevent[2], pevent[3], pevent[4], pevent[5]);
   }
#endif
 
   // these checks are already done before we come here.
   // but we check again as last-ressort protection. K.O.
   assert(total_size > 0);
   assert(total_size >= (int)sizeof(EVENT_HEADER));
 
   rp += total_size;
   if (rp >= pheader->size) {
      rp -= pheader->size;
   } else if ((rp + (int) sizeof(EVENT_HEADER)) > pheader->size) {
      // note: ">" here to match bm_write_to_buffer_locked() and bm_validate_rp().
      // if at the end of the buffer, the remaining free space is exactly
      // equal to the size of an event header, the event header
      // is written there, the pointer is wrapped and the event data
      // is written to the beginning of the buffer.
      rp = 0;
   }
   return rp;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_match_event()

INT bm_match_event	(	short int	event_id,
		short int	trigger_mask,
		const EVENT_HEADER *	pevent
	)

Check if an event matches a given event request by the event id and trigger mask

Parameters

event_id	Event ID of request
trigger_mask	Trigger mask of request
pevent	Pointer to event to check

Returns

TRUE if event matches request

Definition at line 6023 of file midas.cxx.

                                                                                           {
   // NB: cast everything to unsigned 16 bit to avoid bitwise comparison failure
   // because of mismatch in sign-extension between signed 16-bit event_id and
   // unsigned 16-bit constants. K.O.
 
   if (((uint16_t(pevent->event_id) & uint16_t(0xF000)) == uint16_t(EVENTID_FRAG1)) || ((uint16_t(pevent->event_id) & uint16_t(0xF000)) == uint16_t(EVENTID_FRAG)))
      /* fragmented event */
      return (((uint16_t(event_id) == uint16_t(EVENTID_ALL)) || (uint16_t(event_id) == (uint16_t(pevent->event_id) & uint16_t(0x0FFF))))
              && ((uint16_t(trigger_mask) == uint16_t(TRIGGER_ALL)) || ((uint16_t(trigger_mask) & uint16_t(pevent->trigger_mask)))));
 
   return (((uint16_t(event_id) == uint16_t(EVENTID_ALL)) || (uint16_t(event_id) == uint16_t(pevent->event_id)))
           && ((uint16_t(trigger_mask) == uint16_t(TRIGGER_ALL)) || ((uint16_t(trigger_mask) & uint16_t(pevent->trigger_mask)))));
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_next_rp()

static int bm_next_rp ( const char *  who, const BUFFER_HEADER *  pheader, const char *  pdata, int  rp  )

static

Definition at line 6264 of file midas.cxx.

                                                                                                {
   const EVENT_HEADER *pevent = (const EVENT_HEADER *) (pdata + rp);
   int event_size = pevent->data_size + sizeof(EVENT_HEADER);
   int total_size = ALIGN8(event_size);
 
   if (pevent->data_size <= 0 || total_size <= 0 || total_size > pheader->size) {
      cm_msg(MERROR, "bm_next_rp",
             "error: buffer \"%s\" is corrupted: rp %d points to an invalid event: data_size %d, event size %d, total_size %d, buffer read_pointer %d, write_pointer %d, size %d, called from %s",
             pheader->name,
             rp,
             pevent->data_size,
             event_size,
             total_size,
             pheader->read_pointer,
             pheader->write_pointer,
             pheader->size,
             who);
      return -1;
   }
 
   int remaining = 0;
   if (rp < pheader->write_pointer) {
      remaining = pheader->write_pointer - rp;
   } else {
      remaining = pheader->size - rp;
      remaining += pheader->write_pointer;
   }
 
   //printf("bm_next_rp: total_size %d, remaining %d, rp %d, wp %d, size %d\n", total_size, remaining, rp, pheader->write_pointer, pheader->size);
 
   if (total_size > remaining) {
      cm_msg(MERROR, "bm_next_rp",
             "error: buffer \"%s\" is corrupted: rp %d points to an invalid event: data_size %d, event size %d, total_size %d, buffer read_pointer %d, write_pointer %d, size %d, remaining %d, called from %s",
             pheader->name,
             rp,
             pevent->data_size,
             event_size,
             total_size,
             pheader->read_pointer,
             pheader->write_pointer,
             pheader->size,
             remaining,
             who);
      return -1;
   }
 
   rp = bm_incr_rp_no_check(pheader, rp, total_size);
 
   return rp;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_notify_reader_locked()

static void bm_notify_reader_locked ( BUFFER_HEADER *  pheader, BUFFER_CLIENT *  pc, int  old_write_pointer, int  request_id  )

static

Definition at line 9611 of file midas.cxx.

                                                                                                                      {
   if (request_id >= 0) {
      /* if that client has a request and is suspended, wake it up */
      if (pc->read_wait) {
         char str[80];
         sprintf(str, "B %s %d", pheader->name, request_id);
         ss_resume(pc->port, str);
         //printf("bm_notify_reader_locked: buffer [%s] client [%s] request_id %d, port %d, message [%s]\n", pheader->name, pc->name, request_id, pc->port, str);
         //printf("bm_notify_reader_locked: buffer [%s] client [%s] clear read_wait!\n", pheader->name, pc->name);
         pc->read_wait = FALSE;
      }
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_open_buffer()

INT bm_open_buffer	(	const char *	buffer_name,
		INT	buffer_size,
		INT *	buffer_handle
	)

Open an event buffer. Two default buffers are created by the system. The "SYSTEM" buffer is used to exchange events and the "SYSMSG" buffer is used to exchange system messages. The name and size of the event buffers is defined in midas.h as EVENT_BUFFER_NAME and DEFAULT_BUFFER_SIZE. Following example opens the "SYSTEM" buffer, requests events with ID 1 and enters a main loop. Events are then received in process_event()

#include <stdio.h>
#include "midas.h"
void process_event(HNDLE hbuf, HNDLE request_id, EVENT_HEADER *pheader, void *pevent)
{
  printf("Received event #%d\r",
  pheader->serial_number);
}
main()
{
  INT status, request_id;
  HNDLE hbuf;
  status = cm_connect_experiment("pc810", "Sample", "Simple Analyzer", NULL);
  if (status != CM_SUCCESS)
  return 1;
  bm_open_buffer(EVENT_BUFFER_NAME, DEFAULT_BUFFER_SIZE, &hbuf);
  bm_request_event(hbuf, 1, TRIGGER_ALL, GET_ALL, request_id, process_event);
 
  do
  {
   status = cm_yield(1000);
  } while (status != RPC_SHUTDOWN && status != SS_ABORT);
  cm_disconnect_experiment();
  return 0;
}

Parameters

buffer_name	Name of buffer
buffer_size	Default size of buffer in bytes. Can by overwritten with ODB value
buffer_handle	Buffer handle returned by function

Returns

BM_SUCCESS, BM_CREATED
BM_NO_SHM Shared memory cannot be created
BM_NO_SEMAPHORE Semaphore cannot be created
BM_NO_MEMORY Not enough memory to create buffer descriptor
BM_MEMSIZE_MISMATCH Buffer size conflicts with an existing buffer of different size
BM_INVALID_PARAM Invalid parameter

Definition at line 6725 of file midas.cxx.

                                                                                 {
   INT status;
 
   if (rpc_is_remote()) {
      status = rpc_call(RPC_BM_OPEN_BUFFER, buffer_name, buffer_size, buffer_handle);
 
      HNDLE hDB;
      status = cm_get_experiment_database(&hDB, NULL);
      if (status != SUCCESS || hDB == 0) {
         cm_msg(MERROR, "bm_open_buffer", "cannot open buffer \'%s\' - not connected to ODB", buffer_name);
         return BM_NO_SHM;
      }
 
      _bm_max_event_size = DEFAULT_MAX_EVENT_SIZE;
 
      int size = sizeof(INT);
      status = db_get_value(hDB, 0, "/Experiment/MAX_EVENT_SIZE", &_bm_max_event_size, &size, TID_UINT32, TRUE);
 
      if (status != DB_SUCCESS) {
         cm_msg(MERROR, "bm_open_buffer", "Cannot get ODB /Experiment/MAX_EVENT_SIZE, db_get_value() status %d",
                status);
         return status;
      }
 
      return status;
   }
#ifdef LOCAL_ROUTINES
   {
      HNDLE shm_handle;
      size_t shm_size;
      HNDLE hDB;
      const int max_buffer_size = 2 * 1000 * 1024 * 1024; // limited by 32-bit integers in the buffer header
 
      bm_cleanup("bm_open_buffer", ss_millitime(), FALSE);
 
      if (!buffer_name || !buffer_name[0]) {
         cm_msg(MERROR, "bm_open_buffer", "cannot open buffer with zero name");
         return BM_INVALID_PARAM;
      }
 
      if (strlen(buffer_name) >= NAME_LENGTH) {
         cm_msg(MERROR, "bm_open_buffer", "buffer name \"%s\" is longer than %d bytes", buffer_name, NAME_LENGTH);
         return BM_INVALID_PARAM;
      }
 
      status = cm_get_experiment_database(&hDB, NULL);
 
      if (status != SUCCESS || hDB == 0) {
         //cm_msg(MERROR, "bm_open_buffer", "cannot open buffer \'%s\' - not connected to ODB", buffer_name);
         return BM_NO_SHM;
      }
 
      /* get buffer size from ODB, user parameter as default if not present in ODB */
      std::string odb_path;
      odb_path += "/Experiment/Buffer sizes/";
      odb_path += buffer_name;
 
      int size = sizeof(INT);
      status = db_get_value(hDB, 0, odb_path.c_str(), &buffer_size, &size, TID_UINT32, TRUE);
 
      if (buffer_size <= 0 || buffer_size > max_buffer_size) {
         cm_msg(MERROR, "bm_open_buffer",
                "Cannot open buffer \"%s\", invalid buffer size %d in ODB \"%s\", maximum buffer size is %d",
                buffer_name, buffer_size, odb_path.c_str(), max_buffer_size);
         return BM_INVALID_PARAM;
      }
 
      _bm_max_event_size = DEFAULT_MAX_EVENT_SIZE;
 
      size = sizeof(INT);
      status = db_get_value(hDB, 0, "/Experiment/MAX_EVENT_SIZE", &_bm_max_event_size, &size, TID_UINT32, TRUE);
 
      if (status != DB_SUCCESS) {
         cm_msg(MERROR, "bm_open_buffer", "Cannot get ODB /Experiment/MAX_EVENT_SIZE, db_get_value() status %d",
                status);
         return status;
      }
 
      /* check if buffer already is open */
      gBuffersMutex.lock();
      for (size_t i = 0; i < gBuffers.size(); i++) {
         BUFFER* pbuf = gBuffers[i];
         if (pbuf && pbuf->attached && equal_ustring(pbuf->buffer_name, buffer_name)) {
            *buffer_handle = i + 1;
            gBuffersMutex.unlock();
            return BM_SUCCESS;
         }
      }
      gBuffersMutex.unlock();
 
      // only one thread at a time should create new buffers
 
      static std::mutex gNewBufferMutex;
      std::lock_guard<std::mutex> guard(gNewBufferMutex);
 
      // if we had a race against another thread
      // and while we were waiting for gNewBufferMutex
      // the other thread created this buffer, we return it.
 
      gBuffersMutex.lock();
      for (size_t i = 0; i < gBuffers.size(); i++) {
         BUFFER* pbuf = gBuffers[i];
         if (pbuf && pbuf->attached && equal_ustring(pbuf->buffer_name, buffer_name)) {
            *buffer_handle = i + 1;
            gBuffersMutex.unlock();
            return BM_SUCCESS;
         }
      }
      gBuffersMutex.unlock();
 
      /* allocate new BUFFER object */
 
      BUFFER* pbuf = new BUFFER;
 
      /* there is no constructor for BUFFER object, we have to zero the arrays manually */
 
      for (int i=0; i<MAX_CLIENTS; i++) {
         pbuf->client_count_write_wait[i] = 0;
         pbuf->client_time_write_wait[i] = 0;
      }
 
      /* create buffer semaphore */
 
      status = ss_semaphore_create(buffer_name, &(pbuf->semaphore));
 
      if (status != SS_CREATED && status != SS_SUCCESS) {
         *buffer_handle = 0;
         delete pbuf;
         return BM_NO_SEMAPHORE;
      }
 
      std::string client_name = cm_get_client_name();
 
      /* store client name */
      mstrlcpy(pbuf->client_name, client_name.c_str(), sizeof(pbuf->client_name));
 
      /* store buffer name */
      mstrlcpy(pbuf->buffer_name, buffer_name, sizeof(pbuf->buffer_name));
 
      /* lock buffer semaphore to avoid race with bm_open_buffer() in a different program */
 
      pbuf->attached = true; // required by bm_lock_buffer()
 
      bm_lock_buffer_guard pbuf_guard(pbuf);
 
      if (!pbuf_guard.is_locked()) {
         // cannot happen, no other thread can see this pbuf
         abort();
         return BM_NO_SEMAPHORE;
      }
 
      /* open shared memory */
 
      void *p = NULL;
      status = ss_shm_open(buffer_name, sizeof(BUFFER_HEADER) + buffer_size, &p, &shm_size, &shm_handle, FALSE);
 
      if (status != SS_SUCCESS && status != SS_CREATED) {
         *buffer_handle = 0;
         pbuf_guard.unlock();
         pbuf_guard.invalidate(); // destructor will see a deleted pbuf
         delete pbuf;
         return BM_NO_SHM;
      }
 
      pbuf->buffer_header = (BUFFER_HEADER *) p;
 
      BUFFER_HEADER *pheader = pbuf->buffer_header;
 
      bool shm_created = (status == SS_CREATED);
 
      if (shm_created) {
         /* initialize newly created shared memory */
 
         memset(pheader, 0, sizeof(BUFFER_HEADER) + buffer_size);
 
         mstrlcpy(pheader->name, buffer_name, sizeof(pheader->name));
         pheader->size = buffer_size;
 
      } else {
         /* validate existing shared memory */
 
         if (!equal_ustring(pheader->name, buffer_name)) {
            // unlock before calling cm_msg(). if we are SYSMSG, we wil ldeadlock. K.O.
            pbuf_guard.unlock();
            pbuf_guard.invalidate(); // destructor will see a deleted pbuf
            cm_msg(MERROR, "bm_open_buffer",
                   "Buffer \"%s\" is corrupted, mismatch of buffer name in shared memory \"%s\"", buffer_name,
                   pheader->name);
            *buffer_handle = 0;
            delete pbuf;
            return BM_CORRUPTED;
         }
 
         if ((pheader->num_clients < 0) || (pheader->num_clients > MAX_CLIENTS)) {
            // unlock before calling cm_msg(). if we are SYSMSG, we wil ldeadlock. K.O.
            pbuf_guard.unlock();
            pbuf_guard.invalidate(); // destructor will see a deleted pbuf
            cm_msg(MERROR, "bm_open_buffer", "Buffer \"%s\" is corrupted, num_clients %d exceeds MAX_CLIENTS %d",
                   buffer_name, pheader->num_clients, MAX_CLIENTS);
            *buffer_handle = 0;
            delete pbuf;
            return BM_CORRUPTED;
         }
 
         if ((pheader->max_client_index < 0) || (pheader->max_client_index > MAX_CLIENTS)) {
            // unlock before calling cm_msg(). if we are SYSMSG, we wil ldeadlock. K.O.
            pbuf_guard.unlock();
            pbuf_guard.invalidate(); // destructor will see a deleted pbuf
            cm_msg(MERROR, "bm_open_buffer", "Buffer \"%s\" is corrupted, max_client_index %d exceeds MAX_CLIENTS %d",
                   buffer_name, pheader->max_client_index, MAX_CLIENTS);
            *buffer_handle = 0;
            delete pbuf;
            return BM_CORRUPTED;
         }
 
         /* check if buffer size is identical */
         if (pheader->size != buffer_size) {
            cm_msg(MINFO, "bm_open_buffer", "Buffer \"%s\" requested size %d differs from existing size %d",
                   buffer_name, buffer_size, pheader->size);
 
            buffer_size = pheader->size;
 
            ss_shm_close(buffer_name, p, shm_size, shm_handle, FALSE);
 
            status = ss_shm_open(buffer_name, sizeof(BUFFER_HEADER) + buffer_size, &p, &shm_size, &shm_handle, FALSE);
 
            if (status != SS_SUCCESS) {
               *buffer_handle = 0;
               pbuf_guard.unlock();
               pbuf_guard.invalidate(); // destructor will see a deleted pbuf
               delete pbuf;
               return BM_NO_SHM;
            }
 
            pbuf->buffer_header = (BUFFER_HEADER *) p;
            pheader = pbuf->buffer_header;
         }
      }
 
      /* shared memory is good from here down */
 
      pbuf->attached = true;
 
      pbuf->shm_handle = shm_handle;
      pbuf->shm_size = shm_size;
      pbuf->callback = FALSE;
 
      bm_cleanup_buffer_locked(pbuf, "bm_open_buffer", ss_millitime());
 
      status = bm_validate_buffer_locked(pbuf);
      if (status != BM_SUCCESS) {
         cm_msg(MERROR, "bm_open_buffer",
                "buffer \'%s\' is corrupted, bm_validate_buffer() status %d, calling bm_reset_buffer()...", buffer_name,
                status);
         bm_reset_buffer_locked(pbuf);
         cm_msg(MINFO, "bm_open_buffer", "buffer \'%s\' was reset, all buffered events were lost", buffer_name);
      }
 
      /* add our client BUFFER_HEADER */
 
      int iclient = 0;
      for (; iclient < MAX_CLIENTS; iclient++)
         if (pheader->client[iclient].pid == 0)
            break;
 
      if (iclient == MAX_CLIENTS) {
         *buffer_handle = 0;
         // unlock before calling cm_msg(). if we are SYSMSG, we wil ldeadlock. K.O.
         pbuf_guard.unlock();
         pbuf_guard.invalidate(); // destructor will see a deleted pbuf
         delete pbuf;
         cm_msg(MERROR, "bm_open_buffer", "buffer \'%s\' maximum number of clients %d exceeded", buffer_name, MAX_CLIENTS);
         return BM_NO_SLOT;
      }
 
      /* store slot index in _buffer structure */
      pbuf->client_index = iclient;
 
      /*
         Save the index of the last client of that buffer so that later only
         the clients 0..max_client_index-1 have to be searched through.
       */
      pheader->num_clients++;
      if (iclient + 1 > pheader->max_client_index)
         pheader->max_client_index = iclient + 1;
 
      /* setup buffer header and client structure */
      BUFFER_CLIENT *pclient = &pheader->client[iclient];
 
      memset(pclient, 0, sizeof(BUFFER_CLIENT));
 
      mstrlcpy(pclient->name, client_name.c_str(), sizeof(pclient->name));
 
      pclient->pid = ss_getpid();
 
      ss_suspend_get_buffer_port(ss_gettid(), &pclient->port);
 
      pclient->read_pointer = pheader->write_pointer;
      pclient->last_activity = ss_millitime();
 
      cm_get_watchdog_params(NULL, &pclient->watchdog_timeout);
 
      pbuf_guard.unlock();
 
      /* shared memory is not locked from here down, do not touch pheader and pbuf->buffer_header! */
 
      pheader = NULL;
 
      /* we are not holding any locks from here down, but other threads cannot see this pbuf yet */
 
      bm_clear_buffer_statistics(hDB, pbuf);
      bm_write_buffer_statistics_to_odb(hDB, pbuf, true);
 
      /* add pbuf to buffer list */
 
      gBuffersMutex.lock();
 
      bool added = false;
      for (size_t i=0; i<gBuffers.size(); i++) {
         if (gBuffers[i] == NULL) {
            gBuffers[i] = pbuf;
            added = true;
            *buffer_handle = i+1;
            break;
         }
      }
      if (!added) {
         *buffer_handle = gBuffers.size() + 1;
         gBuffers.push_back(pbuf);
      }
 
      /* from here down we should not touch pbuf without locking it */
 
      pbuf = NULL;
 
      gBuffersMutex.unlock();
 
      /* new buffer is now ready for use */
 
      /* initialize buffer counters */
      bm_init_buffer_counters(*buffer_handle);
 
      bm_cleanup("bm_open_buffer", ss_millitime(), FALSE);
 
      if (shm_created)
         return BM_CREATED;
   }
#endif                          /* LOCAL_ROUTINES */
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_peek_buffer_locked()

static int bm_peek_buffer_locked ( BUFFER *  pbuf, BUFFER_HEADER *  pheader, BUFFER_CLIENT *  pc, EVENT_HEADER **  ppevent, int *  pevent_size, int *  ptotal_size  )

static

Definition at line 8899 of file midas.cxx.

{
   if (pc->read_pointer == pheader->write_pointer) {
      /* no more events buffered for this client */
      if (!pc->read_wait) {
         //printf("bm_peek_buffer_locked: buffer [%s] client [%s], set read_wait!\n", pheader->name, pc->name);
         pc->read_wait = TRUE;
      }
      return BM_ASYNC_RETURN;
   }
 
   if (pc->read_wait) {
      //printf("bm_peek_buffer_locked: buffer [%s] client [%s], clear read_wait!\n", pheader->name, pc->name);
      pc->read_wait = FALSE;
   }
 
   if ((pc->read_pointer < 0) || (pc->read_pointer >= pheader->size)) {
      cm_msg(MERROR, "bm_peek_buffer_locked", "event buffer \"%s\" is corrupted: client \"%s\" read pointer %d is invalid. buffer read pointer %d, write pointer %d, size %d", pheader->name, pc->name, pc->read_pointer, pheader->read_pointer, pheader->write_pointer, pheader->size);
      return BM_CORRUPTED;
   }
 
   char *pdata = (char *) (pheader + 1);
 
   EVENT_HEADER *pevent = (EVENT_HEADER *) (pdata + pc->read_pointer);
   int event_size = pevent->data_size + sizeof(EVENT_HEADER);
   int total_size = ALIGN8(event_size);
 
   if ((total_size <= 0) || (total_size > pheader->size)) {
      cm_msg(MERROR, "bm_peek_buffer_locked", "event buffer \"%s\" is corrupted: client \"%s\" read pointer %d points to invalid event: data_size %d, event_size %d, total_size %d. buffer size: %d, read_pointer: %d, write_pointer: %d", pheader->name, pc->name, pc->read_pointer, pevent->data_size, event_size, total_size, pheader->size, pheader->read_pointer, pheader->write_pointer);
      return BM_CORRUPTED;
   }
 
   assert(total_size > 0);
   assert(total_size <= pheader->size);
 
   if (ppevent)
      *ppevent = pevent;
   if (pevent_size)
      *pevent_size = event_size;
   if (ptotal_size)
      *ptotal_size = total_size;
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_peek_read_cache_locked()

static BOOL bm_peek_read_cache_locked ( BUFFER *  pbuf, EVENT_HEADER **  ppevent, int *  pevent_size, int *  ptotal_size  )

static

Definition at line 8873 of file midas.cxx.

{
   if (pbuf->read_cache_rp == pbuf->read_cache_wp)
      return FALSE;
 
   EVENT_HEADER *pevent = (EVENT_HEADER *) (pbuf->read_cache + pbuf->read_cache_rp);
   int event_size = pevent->data_size + sizeof(EVENT_HEADER);
   int total_size = ALIGN8(event_size);
 
   if (ppevent)
      *ppevent = pevent;
   if (pevent_size)
      *pevent_size = event_size;
   if (ptotal_size)
      *ptotal_size = total_size;
 
   return TRUE;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_poll_event()

INT bm_poll_event

(

void

)

Definition at line 11134 of file midas.cxx.

{
   BOOL dispatched_something = FALSE;
 
   //printf("bm_poll_event!\n");
 
   DWORD start_time = ss_millitime();
 
   std::vector<char> vec;
 
   /* loop over all requests */
   _request_list_mutex.lock();
   bool locked = true;
   size_t n = _request_list.size();
   for (size_t i = 0; i < n; i++) {
      if (!locked) {
         _request_list_mutex.lock();
         locked = true;
      }
      /* continue if no dispatcher set (manual bm_receive_event) */
      if (_request_list[i].dispatcher == NULL)
         continue;
 
      int buffer_handle = _request_list[i].buffer_handle;
 
      /* must release the lock on the request list: user provided r.dispatcher() can add or remove event requests, and we will deadlock. K.O. */
      _request_list_mutex.unlock();
      locked = false;
 
      do {
         /* receive event */
         int status = bm_receive_event_vec(buffer_handle, &vec, BM_NO_WAIT);
 
         //printf("bm_poll_event: request_id %d, buffer_handle %d, bm_receive_event(BM_NO_WAIT) status %d, vec size %d, capacity %d\n", request_id, buffer_handle, status, (int)vec.size(), (int)vec.capacity());
 
         /* call user function if successful */
         if (status == BM_SUCCESS) {
            bm_dispatch_event(buffer_handle, (EVENT_HEADER*)vec.data());
            dispatched_something = TRUE;
         }
 
         /* break if no more events */
         if (status == BM_ASYNC_RETURN)
            break;
 
         /* break if corrupted event buffer */
         if (status == BM_TRUNCATED) {
            cm_msg(MERROR, "bm_poll_event", "received event was truncated, buffer size %d is too small, see messages and increase /Experiment/MAX_EVENT_SIZE in ODB", (int)vec.size());
         }
 
         /* break if corrupted event buffer */
         if (status == BM_CORRUPTED)
            return SS_ABORT;
 
         /* break if server died */
         if (status == RPC_NET_ERROR) {
            return SS_ABORT;
         }
 
         /* stop after one second */
         if (ss_millitime() - start_time > 1000) {
            break;
         }
 
      } while (TRUE);
   }
 
   if (locked)
      _request_list_mutex.unlock();
 
   if (dispatched_something)
      return BM_SUCCESS;
   else
      return BM_ASYNC_RETURN;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_push_buffer()

static INT bm_push_buffer ( BUFFER *  pbuf, int  buffer_handle  )

static

Check a buffer if an event is available and call the dispatch function if found.

Parameters

buffer_name

Name of buffer

Returns

BM_SUCCESS, BM_INVALID_HANDLE, BM_TRUNCATED, BM_ASYNC_RETURN, RPC_NET_ERROR

Definition at line 10910 of file midas.cxx.

                                                           {
   //printf("bm_push_buffer: buffer [%s], handle %d, callback %d\n", pbuf->buffer_header->name, buffer_handle, pbuf->callback);
 
   /* return immediately if no callback routine is defined */
   if (!pbuf->callback)
      return BM_SUCCESS;
 
   return bm_read_buffer(pbuf, buffer_handle, NULL, NULL, NULL, NULL, BM_NO_WAIT, 0, TRUE);
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_read_buffer()

static INT bm_read_buffer ( BUFFER *  pbuf, INT  buffer_handle, void **  bufptr, void *  buf, INT *  buf_size, std::vector< char > *  vecptr, int  timeout_msec, int  convert_flags, BOOL  dispatch  )

static

Definition at line 10272 of file midas.cxx.

                                                                                                                                                                                 {
   INT status = BM_SUCCESS;
 
   int max_size = 0;
   if (buf_size) {
      max_size = *buf_size;
      *buf_size = 0;
   }
 
   //printf("bm_read_buffer: [%s] timeout %d, conv %d, ptr %p, buf %p, disp %d\n", pbuf->buffer_name, timeout_msec, convert_flags, bufptr, buf, dispatch);
 
   bm_lock_buffer_guard pbuf_guard(pbuf, true); // buffer is not locked
 
   // NB: locking order is: 1st read cache lock, 2nd buffer lock, unlock in reverse order
 
   /* look if there is anything in the cache */
   if (pbuf->read_cache_size > 0) {
 
      status = bm_lock_buffer_read_cache(pbuf);
 
      if (status != BM_SUCCESS)
         return status;
 
      if (pbuf->read_cache_wp == 0) {
 
         // lock buffer for the first time
 
         if (!pbuf_guard.relock()) {
            pbuf->read_cache_mutex.unlock();
            return pbuf_guard.get_status();
         }
 
         status = bm_fill_read_cache_locked(pbuf_guard, timeout_msec);
         if (status != BM_SUCCESS) {
            // unlock in correct order
            if (pbuf_guard.is_locked()) {
               // check if bm_wait_for_more_events() failed to relock the buffer
               pbuf_guard.unlock();
            }
            pbuf->read_cache_mutex.unlock();
            return status;
         }
 
         // buffer remains locked here
      }
      EVENT_HEADER *pevent;
      int event_size;
      int total_size;
      if (bm_peek_read_cache_locked(pbuf, &pevent, &event_size, &total_size)) {
         if (pbuf_guard.is_locked()) {
            // do not need to keep the event buffer locked
            // when reading from the read cache
            pbuf_guard.unlock();
         }
         //printf("bm_read_buffer: [%s] async %d, conv %d, ptr %p, buf %p, disp %d, total_size %d, read from cache %d %d %d\n", pbuf->buffer_name, async_flag, convert_flags, bufptr, buf, dispatch, total_size, pbuf->read_cache_size, pbuf->read_cache_rp, pbuf->read_cache_wp);
         status = BM_SUCCESS;
         if (buf) {
            if (event_size > max_size) {
               cm_msg(MERROR, "bm_read_buffer", "buffer size %d is smaller than event size %d, event truncated. buffer \"%s\"", max_size, event_size, pbuf->buffer_name);
               event_size = max_size;
               status = BM_TRUNCATED;
            }
 
            memcpy(buf, pevent, event_size);
 
            if (buf_size) {
               *buf_size = event_size;
            }
            if (convert_flags) {
               bm_convert_event_header((EVENT_HEADER *) buf, convert_flags);
            }
         } else if (bufptr) {
            *bufptr = malloc(event_size);
            memcpy(*bufptr, pevent, event_size);
            status = BM_SUCCESS;
         } else if (vecptr) {
            vecptr->resize(0);
            char* cptr = (char*)pevent;
            vecptr->assign(cptr, cptr+event_size);
         }
         bm_incr_read_cache_locked(pbuf, total_size);
         pbuf->read_cache_mutex.unlock();
         if (dispatch) {
            // FIXME need to protect currently dispatched event against
            // another thread overwriting it by refilling the read cache
            bm_dispatch_event(buffer_handle, pevent);
            return BM_MORE_EVENTS;
         }
         // buffer is unlocked here
         return status;
      }
      pbuf->read_cache_mutex.unlock();
   }
 
   /* we come here if the read cache is disabled */
   /* we come here if the next event is too big to fit into the read cache */
 
   if (!pbuf_guard.is_locked()) {
      if (!pbuf_guard.relock())
         return pbuf_guard.get_status();
   }
 
   EVENT_HEADER *event_buffer = NULL;
 
   BUFFER_HEADER *pheader = pbuf->buffer_header;
 
   BUFFER_CLIENT *pc = bm_get_my_client_locked(pbuf_guard);
 
   while (1) {
      /* loop over events in the event buffer */
 
      status = bm_wait_for_more_events_locked(pbuf_guard, pc, timeout_msec, FALSE);
 
      if (status != BM_SUCCESS) {
         // implicit unlock
         return status;
      }
 
      /* check if event at current read pointer matches a request */
 
      EVENT_HEADER *pevent;
      int event_size;
      int total_size;
 
      status = bm_peek_buffer_locked(pbuf, pheader, pc, &pevent, &event_size, &total_size);
      if (status == BM_CORRUPTED) {
         // implicit unlock
         return status;
      } else if (status != BM_SUCCESS) {
         /* event buffer is empty */
         break;
      }
 
      BOOL is_requested = bm_check_requests(pc, pevent);
 
      if (is_requested) {
         //printf("bm_read_buffer: [%s] async %d, conv %d, ptr %p, buf %p, disp %d, total_size %d, read from buffer, cache %d %d %d\n", pheader->name, async_flag, convert_flags, bufptr, buf, dispatch, total_size, pbuf->read_cache_size, pbuf->read_cache_rp, pbuf->read_cache_wp);
 
         status = BM_SUCCESS;
 
         if (buf) {
            if (event_size > max_size) {
               cm_msg(MERROR, "bm_read_buffer",
                      "buffer size %d is smaller than event size %d, event truncated. buffer \"%s\"", max_size,
                      event_size, pheader->name);
               event_size = max_size;
               status = BM_TRUNCATED;
            }
 
            bm_read_from_buffer_locked(pheader, pc->read_pointer, (char *) buf, event_size);
 
            if (buf_size) {
               *buf_size = event_size;
            }
 
            if (convert_flags) {
               bm_convert_event_header((EVENT_HEADER *) buf, convert_flags);
            }
 
            pbuf->count_read++;
            pbuf->bytes_read += event_size;
         } else if (dispatch || bufptr) {
            assert(event_buffer == NULL); // make sure we only come here once
            event_buffer = (EVENT_HEADER *) malloc(event_size);
            bm_read_from_buffer_locked(pheader, pc->read_pointer, (char *) event_buffer, event_size);
            pbuf->count_read++;
            pbuf->bytes_read += event_size;
         } else if (vecptr) {
            bm_read_from_buffer_locked(pheader, pc->read_pointer, vecptr, event_size);
            pbuf->count_read++;
            pbuf->bytes_read += event_size;
         }
 
         int new_read_pointer = bm_incr_rp_no_check(pheader, pc->read_pointer, total_size);
         pc->read_pointer = new_read_pointer;
 
         pheader->num_out_events++;
         /* exit loop over events */
         break;
      }
 
      int new_read_pointer = bm_incr_rp_no_check(pheader, pc->read_pointer, total_size);
      pc->read_pointer = new_read_pointer;
      pheader->num_out_events++;
   }
 
   /*
     If read pointer has been changed, it may have freed up some space
     for waiting producers. So check if free space is now more than 50%
     of the buffer size and wake waiting producers.
   */
 
   bm_wakeup_producers_locked(pheader, pc);
 
   pbuf_guard.unlock();
 
   if (dispatch && event_buffer) {
      bm_dispatch_event(buffer_handle, event_buffer);
      free(event_buffer);
      event_buffer = NULL;
      return BM_MORE_EVENTS;
   }
 
   if (bufptr && event_buffer) {
      *bufptr = event_buffer;
      event_buffer = NULL;
      status = BM_SUCCESS;
   }
 
   if (event_buffer) {
      free(event_buffer);
      event_buffer = NULL;
   }
 
   return status;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_read_from_buffer_locked() [1/2]

static void bm_read_from_buffer_locked ( const BUFFER_HEADER *  pheader, int  rp, char *  buf, int  event_size  )

static

Definition at line 8944 of file midas.cxx.

{
   const char *pdata = (const char *) (pheader + 1);
 
   if (rp + event_size <= pheader->size) {
      /* copy event to cache */
      memcpy(buf, pdata + rp, event_size);
   } else {
      /* event is splitted */
      int size = pheader->size - rp;
      memcpy(buf, pdata + rp, size);
      memcpy(buf + size, pdata, event_size - size);
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_read_from_buffer_locked() [2/2]

static void bm_read_from_buffer_locked ( const BUFFER_HEADER *  pheader, int  rp, std::vector< char > *  vecptr, int  event_size  )

static

Definition at line 8959 of file midas.cxx.

{
   const char *pdata = (const char *) (pheader + 1);
 
   if (rp + event_size <= pheader->size) {
      /* copy event to cache */
      vecptr->assign(pdata + rp, pdata + rp + event_size);
   } else {
      /* event is splitted */
      int size = pheader->size - rp;
      vecptr->assign(pdata + rp, pdata + rp + size);
      vecptr->insert(vecptr->end(), pdata, pdata + event_size - size);
   }
}

Here is the call graph for this function:

bm_receive_event()

INT bm_receive_event	(	INT	buffer_handle,
		void *	destination,
		INT *	buf_size,
		int	timeout_msec
	)

Receives events directly. This function is an alternative way to receive events without a main loop.

It can be used in analysis systems which actively receive events, rather than using callbacks. A analysis package could for example contain its own command line interface. A command like "receive 1000 events" could make it necessary to call bm_receive_event() 1000 times in a row to receive these events and then return back to the command line prompt. The according bm_request_event() call contains NULL as the callback routine to indicate that bm_receive_event() is called to receive events.

#include <stdio.h>
#include "midas.h"
void process_event(EVENT_HEADER *pheader)
{
 printf("Received event #%d\r",
 pheader->serial_number);
}
main()
{
  INT status, request_id;
  HNDLE hbuf;
  char event_buffer[1000];
  status = cm_connect_experiment("", "Sample",
  "Simple Analyzer", NULL);
  if (status != CM_SUCCESS)
   return 1;
  bm_open_buffer(EVENT_BUFFER_NAME, DEFAULT_BUFFER_SIZE, &hbuf);
  bm_request_event(hbuf, 1, TRIGGER_ALL, GET_ALL, request_id, NULL);
 
  do
  {
   size = sizeof(event_buffer);
   status = bm_receive_event(hbuf, event_buffer, &size, BM_NO_WAIT);
  if (status == CM_SUCCESS)
   process_event((EVENT_HEADER *) event_buffer);
   <...do something else...>
   status = cm_yield(0);
  } while (status != RPC_SHUTDOWN &&
  status != SS_ABORT);
  cm_disconnect_experiment();
  return 0;
}

Parameters

buffer_handle	buffer handle
destination	destination address where event is written to
buf_size	size of destination buffer on input, size of event plus header on return.
timeout_msec	Wait so many millisecond for new data. Special values: BM_WAIT: wait forever, BM_NO_WAIT: do not wait, return BM_ASYNC_RETURN if no data is immediately available

Returns

BM_SUCCESS, BM_INVALID_HANDLE
BM_TRUNCATED The event is larger than the destination buffer and was therefore truncated
BM_ASYNC_RETURN No event available

Definition at line 10658 of file midas.cxx.

                                                                                            {
   //printf("bm_receive_event: handle %d, async %d\n", buffer_handle, async_flag);
   if (rpc_is_remote()) {
      return bm_receive_event_rpc(buffer_handle, destination, buf_size, NULL, NULL, timeout_msec);
   }
#ifdef LOCAL_ROUTINES
   {
      INT status = BM_SUCCESS;
 
      BUFFER *pbuf = bm_get_buffer("bm_receive_event", buffer_handle, &status);
 
      if (!pbuf)
         return status;
 
      int convert_flags = rpc_get_convert_flags();
 
      status = bm_read_buffer(pbuf, buffer_handle, NULL, destination, buf_size, NULL, timeout_msec, convert_flags, FALSE);
      //printf("bm_receive_event: handle %d, async %d, status %d, size %d\n", buffer_handle, async_flag, status, *buf_size);
      return status;
   }
#else                           /* LOCAL_ROUTINES */
 
   return BM_SUCCESS;
#endif
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_receive_event_alloc()

INT bm_receive_event_alloc	(	INT	buffer_handle,
		EVENT_HEADER **	ppevent,
		int	timeout_msec
	)

Receives events directly. This function is an alternative way to receive events without a main loop.

It can be used in analysis systems which actively receive events, rather than using callbacks. A analysis package could for example contain its own command line interface. A command like "receive 1000 events" could make it necessary to call bm_receive_event() 1000 times in a row to receive these events and then return back to the command line prompt. The according bm_request_event() call contains NULL as the callback routine to indicate that bm_receive_event() is called to receive events.

#include <stdio.h>
#include "midas.h"
void process_event(EVENT_HEADER *pheader)
{
 printf("Received event #%d\r",
 pheader->serial_number);
}
main()
{
  INT status, request_id;
  HNDLE hbuf;
  char event_buffer[1000];
  status = cm_connect_experiment("", "Sample",
  "Simple Analyzer", NULL);
  if (status != CM_SUCCESS)
   return 1;
  bm_open_buffer(EVENT_BUFFER_NAME, DEFAULT_BUFFER_SIZE, &hbuf);
  bm_request_event(hbuf, 1, TRIGGER_ALL, GET_ALL, request_id, NULL);
 
  do
  {
   size = sizeof(event_buffer);
   status = bm_receive_event(hbuf, event_buffer, &size, BM_NO_WAIT);
  if (status == CM_SUCCESS)
   process_event((EVENT_HEADER *) event_buffer);
   <...do something else...>
   status = cm_yield(0);
  } while (status != RPC_SHUTDOWN &&
  status != SS_ABORT);
  cm_disconnect_experiment();
  return 0;
}

Parameters

buffer_handle	buffer handle
ppevent	pointer to the received event pointer, event pointer should be free()ed to avoid memory leak
timeout_msec	Wait so many millisecond for new data. Special values: BM_WAIT: wait forever, BM_NO_WAIT: do not wait, return BM_ASYNC_RETURN if no data is immediately available

Returns

BM_SUCCESS, BM_INVALID_HANDLE
BM_ASYNC_RETURN No event available

Definition at line 10739 of file midas.cxx.

                                                                                        {
   if (rpc_is_remote()) {
      return bm_receive_event_rpc(buffer_handle, NULL, NULL, ppevent, NULL, timeout_msec);
   }
#ifdef LOCAL_ROUTINES
   {
      INT status = BM_SUCCESS;
 
      BUFFER *pbuf = bm_get_buffer("bm_receive_event_alloc", buffer_handle, &status);
 
      if (!pbuf)
         return status;
 
      int convert_flags = rpc_get_convert_flags();
 
      return bm_read_buffer(pbuf, buffer_handle, (void **) ppevent, NULL, NULL, NULL, timeout_msec, convert_flags, FALSE);
   }
#else                           /* LOCAL_ROUTINES */
 
   return BM_SUCCESS;
#endif
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_receive_event_rpc()

static INT bm_receive_event_rpc ( INT  buffer_handle, void *  buf, int *  buf_size, EVENT_HEADER **  ppevent, std::vector< char > *  pvec, int  timeout_msec  )

static

Definition at line 10491 of file midas.cxx.

{
   //printf("bm_receive_event_rpc: handle %d, buf %p, pevent %p, pvec %p, timeout %d, max_event_size %d\n", buffer_handle, buf, ppevent, pvec, timeout_msec, _bm_max_event_size);
 
   assert(_bm_max_event_size > sizeof(EVENT_HEADER));
 
   void *xbuf = NULL;
   int xbuf_size = 0;
 
   if (buf) {
      xbuf = buf;
      xbuf_size = *buf_size;
   } else if (ppevent) {
      *ppevent = (EVENT_HEADER*)malloc(_bm_max_event_size);
      xbuf_size = _bm_max_event_size;
   } else if (pvec) {
      pvec->resize(_bm_max_event_size);
      xbuf = pvec->data();
      xbuf_size = pvec->size();
   } else {
      assert(!"incorrect call to bm_receivent_event_rpc()");
   }
 
   int status;
   DWORD time_start = ss_millitime();
   DWORD time_end = time_start + timeout_msec;
   
   int xtimeout_msec = timeout_msec;
 
   int zbuf_size = xbuf_size;
 
   while (1) {
      if (timeout_msec == BM_WAIT) {
         xtimeout_msec = 1000;
      } else if (timeout_msec == BM_NO_WAIT) {
         xtimeout_msec = BM_NO_WAIT;
      } else {
         if (xtimeout_msec > 1000) {
            xtimeout_msec = 1000;
         }
      }
 
      zbuf_size = xbuf_size;
 
      status = rpc_call(RPC_BM_RECEIVE_EVENT, buffer_handle, xbuf, &zbuf_size, xtimeout_msec);
      
      //printf("bm_receive_event_rpc: handle %d, timeout %d, status %d, size %d in, %d out, via RPC_BM_RECEIVE_EVENT\n", buffer_handle, xtimeout_msec, status, xbuf_size, zbuf_size);
 
      if (status == BM_ASYNC_RETURN) {
         if (timeout_msec == BM_WAIT) {
            // BM_WAIT means wait forever
            continue;
         } else if (timeout_msec == BM_NO_WAIT) {
            // BM_NO_WAIT means do not wait
            break;
         } else {
            DWORD now = ss_millitime();
            if (now >= time_end) {
               // timeout, return BM_ASYNC_RETURN
               break;
            }
 
            DWORD remain = time_end - now;
 
            if (remain < (DWORD)xtimeout_msec) {
               xtimeout_msec = remain;
            }
 
            // keep asking for event...
            continue;
         }
      } else if (status == BM_SUCCESS) {
         // success, return BM_SUCCESS
         break;
      }
 
      // RPC error
         
      if (buf) {
         *buf_size = 0;
      } else if (ppevent) {
         free(*ppevent);
         *ppevent = NULL;
      } else if (pvec) {
         pvec->resize(0);
      } else {
         assert(!"incorrect call to bm_receivent_event_rpc()");
      }
      
      return status;
   }
 
   // status is BM_SUCCESS or BM_ASYNC_RETURN
 
   if (buf) {
      *buf_size = zbuf_size;
   } else if (ppevent) {
      // nothing to do
      // ppevent = realloc(ppevent, xbuf_size); // shrink memory allocation
   } else if (pvec) {
      pvec->resize(zbuf_size);
   } else {
      assert(!"incorrect call to bm_receivent_event_rpc()");
   }
 
   return status;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_receive_event_vec()

INT bm_receive_event_vec	(	INT	buffer_handle,
		std::vector< char > *	pvec,
		int	timeout_msec
	)

Receives events directly. This function is an alternative way to receive events without a main loop.

It can be used in analysis systems which actively receive events, rather than using callbacks. A analysis package could for example contain its own command line interface. A command like "receive 1000 events" could make it necessary to call bm_receive_event() 1000 times in a row to receive these events and then return back to the command line prompt. The according bm_request_event() call contains NULL as the callback routine to indicate that bm_receive_event() is called to receive events.

#include <stdio.h>
#include "midas.h"
void process_event(EVENT_HEADER *pheader)
{
 printf("Received event #%d\r",
 pheader->serial_number);
}
main()
{
  INT status, request_id;
  HNDLE hbuf;
  char event_buffer[1000];
  status = cm_connect_experiment("", "Sample",
  "Simple Analyzer", NULL);
  if (status != CM_SUCCESS)
   return 1;
  bm_open_buffer(EVENT_BUFFER_NAME, DEFAULT_BUFFER_SIZE, &hbuf);
  bm_request_event(hbuf, 1, TRIGGER_ALL, GET_ALL, request_id, NULL);
 
  do
  {
   size = sizeof(event_buffer);
   status = bm_receive_event(hbuf, event_buffer, &size, BM_NO_WAIT);
  if (status == CM_SUCCESS)
   process_event((EVENT_HEADER *) event_buffer);
   <...do something else...>
   status = cm_yield(0);
  } while (status != RPC_SHUTDOWN &&
  status != SS_ABORT);
  cm_disconnect_experiment();
  return 0;
}

Parameters

buffer_handle	buffer handle
ppevent	pointer to the received event pointer, event pointer should be free()ed to avoid memory leak
timeout_msec	Wait so many millisecond for new data. Special values: BM_WAIT: wait forever, BM_NO_WAIT: do not wait, return BM_ASYNC_RETURN if no data is immediately available

Returns

BM_SUCCESS, BM_INVALID_HANDLE
BM_ASYNC_RETURN No event available

Definition at line 10817 of file midas.cxx.

                                                                                     {
   if (rpc_is_remote()) {
      return bm_receive_event_rpc(buffer_handle, NULL, NULL, NULL, pvec, timeout_msec);
   }
#ifdef LOCAL_ROUTINES
   {
      INT status = BM_SUCCESS;
 
      BUFFER *pbuf = bm_get_buffer("bm_receive_event_vec", buffer_handle, &status);
 
      if (!pbuf)
         return status;
 
      int convert_flags = rpc_get_convert_flags();
 
      return bm_read_buffer(pbuf, buffer_handle, NULL, NULL, NULL, pvec, timeout_msec, convert_flags, FALSE);
   }
#else /* LOCAL_ROUTINES */
   return BM_SUCCESS;
#endif
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_remove_client_locked()

void bm_remove_client_locked	(	BUFFER_HEADER *	pheader,
		int	j
	)

Called to forcibly disconnect given client from a data buffer

Definition at line 6043 of file midas.cxx.

                                                            {
   int k, nc;
   BUFFER_CLIENT *pbctmp;
 
   /* clear entry from client structure in buffer header */
   memset(&(pheader->client[j]), 0, sizeof(BUFFER_CLIENT));
 
   /* calculate new max_client_index entry */
   for (k = MAX_CLIENTS - 1; k >= 0; k--)
      if (pheader->client[k].pid != 0)
         break;
   pheader->max_client_index = k + 1;
 
   /* count new number of clients */
   for (k = MAX_CLIENTS - 1, nc = 0; k >= 0; k--)
      if (pheader->client[k].pid != 0)
         nc++;
   pheader->num_clients = nc;
 
   /* check if anyone is waiting and wake him up */
   pbctmp = pheader->client;
 
   for (k = 0; k < pheader->max_client_index; k++, pbctmp++)
      if (pbctmp->pid && (pbctmp->write_wait || pbctmp->read_wait))
         ss_resume(pbctmp->port, "B  ");
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_remove_event_request()

INT bm_remove_event_request	(	INT	buffer_handle,
		INT	request_id
	)

Delete a previously placed request for a specific event type in the client structure of the buffer refereced by buffer_handle.

Parameters

buffer_handle	Handle to the buffer where the re- quest should be placed in
request_id	Request id returned by bm_request_event

Returns

BM_SUCCESS, BM_INVALID_HANDLE, BM_NOT_FOUND, RPC_NET_ERROR

Definition at line 8526 of file midas.cxx.

                                                               {
   if (rpc_is_remote())
      return rpc_call(RPC_BM_REMOVE_EVENT_REQUEST, buffer_handle, request_id);
 
#ifdef LOCAL_ROUTINES
   {
      int status = 0;
 
      BUFFER *pbuf = bm_get_buffer("bm_remove_event_request", buffer_handle, &status);
 
      if (!pbuf)
         return status;
 
      /* lock buffer */
      bm_lock_buffer_guard pbuf_guard(pbuf);
 
      if (!pbuf_guard.is_locked())
         return pbuf_guard.get_status();
 
      INT i, deleted;
 
      /* get a pointer to the proper client structure */
      BUFFER_CLIENT *pclient = bm_get_my_client_locked(pbuf_guard);
 
      /* check all requests and set to zero if matching */
      for (i = 0, deleted = 0; i < pclient->max_request_index; i++)
         if (pclient->event_request[i].valid && pclient->event_request[i].id == request_id) {
            memset(&pclient->event_request[i], 0, sizeof(EVENT_REQUEST));
            deleted++;
         }
 
      /* calculate new max_request_index entry */
      for (i = MAX_EVENT_REQUESTS - 1; i >= 0; i--)
         if (pclient->event_request[i].valid)
            break;
 
      pclient->max_request_index = i + 1;
 
      /* calculate new all_flag */
      pclient->all_flag = FALSE;
 
      for (i = 0; i < pclient->max_request_index; i++)
         if (pclient->event_request[i].valid && (pclient->event_request[i].sampling_type & GET_ALL)) {
            pclient->all_flag = TRUE;
            break;
         }
 
      pbuf->get_all_flag = pclient->all_flag;
 
      if (!deleted)
         return BM_NOT_FOUND;
   }
#endif                          /* LOCAL_ROUTINES */
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_request_event()

INT bm_request_event	(	HNDLE	buffer_handle,
		short int	event_id,
		short int	trigger_mask,
		INT	sampling_type,
		HNDLE *	request_id,
		EVENT_HANDLER *	func
	)

dox Place an event request based on certain characteristics. Multiple event requests can be placed for each buffer, which are later identified by their request ID. They can contain different callback routines. Example see bm_open_buffer() and bm_receive_event()

Parameters

buffer_handle	buffer handle obtained via bm_open_buffer()
event_id	event ID for requested events. Use EVENTID_ALL to receive events with any ID.
trigger_mask	trigger mask for requested events. The requested events must have at least one bit in its trigger mask common with the requested trigger mask. Use TRIGGER_ALL to receive events with any trigger mask.
sampling_type	specifies how many events to receive. A value of GET_ALL receives all events which match the specified event ID and trigger mask. If the events are consumed slower than produced, the producer is automatically slowed down. A value of GET_NONBLOCKING receives as much events as possible without slowing down the producer. GET_ALL is typically used by the logger, while GET_NONBLOCKING is typically used by analyzers.
request_id	request ID returned by the function. This ID is passed to the callback routine and must be used in the bm_delete_request() routine.
func	allback routine which gets called when an event of the specified type is received.

Returns

BM_SUCCESS, BM_INVALID_HANDLE
BM_NO_MEMORY too many requests. The value MAX_EVENT_REQUESTS in midas.h should be increased.

Definition at line 8473 of file midas.cxx.

{
   assert(request_id != NULL);
   
   EventRequest r;
   r.buffer_handle = buffer_handle;
   r.event_id      = event_id;
   r.trigger_mask  = trigger_mask;
   r.dispatcher    = func;
 
   {
      std::lock_guard<std::mutex> guard(_request_list_mutex);
   
      bool found = false;
      
      // find deleted entry
      for (size_t i = 0; i < _request_list.size(); i++) {
         if (_request_list[i].buffer_handle == 0) {
            _request_list[i] = r;
            *request_id = i;
            found = true;
            break;
         }
      }
      
      if (!found) { // not found
         *request_id = _request_list.size();
         _request_list.push_back(r);
      }
 
      // implicit unlock()
   }
 
   /* add request in buffer structure */
   int status = bm_add_event_request(buffer_handle, event_id, trigger_mask, sampling_type, func, *request_id);
   if (status != BM_SUCCESS)
      return status;
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_reset_buffer_locked()

static void bm_reset_buffer_locked ( BUFFER *  pbuf )

static

Definition at line 6399 of file midas.cxx.

                                                 {
   BUFFER_HEADER *pheader = pbuf->buffer_header;
 
   //printf("bm_reset_buffer: buffer \"%s\"\n", pheader->name);
 
   pheader->read_pointer = 0;
   pheader->write_pointer = 0;
 
   int i;
   for (i = 0; i < pheader->max_client_index; i++) {
      BUFFER_CLIENT *pc = pheader->client + i;
      if (pc->pid) {
         pc->read_pointer = 0;
      }
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_send_event()

INT bm_send_event	(	INT	buffer_handle,
		const EVENT_HEADER *	pevent,
		int	unused,
		int	timeout_msec
	)

Definition at line 9686 of file midas.cxx.

{
   const DWORD MAX_DATA_SIZE = (0x7FFFFFF0 - 16); // event size computations are not 32-bit clean, limit event size to 2GB. K.O.
   const DWORD data_size = pevent->data_size; // 32-bit unsigned value
 
   if (data_size == 0) {
      cm_msg(MERROR, "bm_send_event", "invalid event data size zero");
      return BM_INVALID_SIZE;
   }
 
   if (data_size > MAX_DATA_SIZE) {
      cm_msg(MERROR, "bm_send_event", "invalid event data size %d (0x%x) maximum is %d (0x%x)", data_size, data_size, MAX_DATA_SIZE, MAX_DATA_SIZE);
      return BM_INVALID_SIZE;
   }
 
   const size_t event_size = sizeof(EVENT_HEADER) + data_size;
 
   //printf("bm_send_event: pevent %p, data_size %d, event_size %d, buf_size %d\n", pevent, data_size, event_size, unused);
 
   if (rpc_is_remote()) {
      //return bm_send_event_rpc(buffer_handle, pevent, event_size, timeout_msec);
      return rpc_send_event_sg(buffer_handle, 1, (char**)&pevent, &event_size);
   } else {
      return bm_send_event_sg(buffer_handle, 1, (char**)&pevent, &event_size, timeout_msec);
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_send_event_sg()

int bm_send_event_sg	(	int	buffer_handle,
		int	sg_n,
		const char *const	sg_ptr[],
		const size_t	sg_len[],
		int	timeout_msec
	)

Sends an event to a buffer. This function check if the buffer has enough space for the event, then copies the event to the buffer in shared memory. If clients have requests for the event, they are notified via an UDP packet.

char event[1000];
// create event with ID 1, trigger mask 0, size 100 bytes and serial number 1
bm_compose_event((EVENT_HEADER *) event, 1, 0, 100, 1);
 
// set first byte of event
*(event+sizeof(EVENT_HEADER)) = <...>
#include <stdio.h>
#include "midas.h"
main()
{
 INT status, i;
 HNDLE hbuf;
 char event[1000];
 status = cm_connect_experiment("", "Sample", "Producer", NULL);
 if (status != CM_SUCCESS)
 return 1;
 bm_open_buffer(EVENT_BUFFER_NAME, DEFAULT_BUFFER_SIZE, &hbuf);
 
 // create event with ID 1, trigger mask 0, size 100 bytes and serial number 1
 bm_compose_event((EVENT_HEADER *) event, 1, 0, 100, 1);
 
 // set event data
 for (i=0 ; i<100 ; i++)
 *(event+sizeof(EVENT_HEADER)+i) = i;
 // send event
 bm_send_event(hbuf, event, 100+sizeof(EVENT_HEADER), BM_WAIT);
 cm_disconnect_experiment();
 return 0;
}

Parameters

buffer_handle	Buffer handle obtained via bm_open_buffer()
source	Address of event buffer
buf_size	Size of event including event header in bytes
timeout_msec	Timeout waiting for free space in the event buffer. If BM_WAIT, wait forever. If BM_NO_WAIT, the function returns immediately with a value of BM_ASYNC_RETURN without writing the event to the buffer

Returns

BM_SUCCESS, BM_INVALID_HANDLE, BM_INVALID_PARAM
BM_ASYNC_RETURN Routine called with timeout_msec == BM_NO_WAIT and buffer has not enough space to receive event
BM_NO_MEMORY Event is too large for network buffer or event buffer. One has to increase the event buffer size "/Experiment/Buffer sizes/SYSTEM" and/or /Experiment/MAX_EVENT_SIZE in ODB.

Definition at line 9786 of file midas.cxx.

{
   if (rpc_is_remote())
      return rpc_send_event_sg(buffer_handle, sg_n, sg_ptr, sg_len);
 
   if (sg_n < 1) {
      cm_msg(MERROR, "bm_send_event", "invalid sg_n %d", sg_n);
      return BM_INVALID_SIZE;
   }
 
   if (sg_ptr[0] == NULL) {
      cm_msg(MERROR, "bm_send_event", "invalid sg_ptr[0] is NULL");
      return BM_INVALID_SIZE;
   }
 
   if (sg_len[0] < sizeof(EVENT_HEADER)) {
      cm_msg(MERROR, "bm_send_event", "invalid sg_len[0] value %d is smaller than event header size %d", (int)sg_len[0], (int)sizeof(EVENT_HEADER));
      return BM_INVALID_SIZE;
   }
 
   const EVENT_HEADER* pevent = (const EVENT_HEADER*)sg_ptr[0];
   
   const DWORD MAX_DATA_SIZE = (0x7FFFFFF0 - 16); // event size computations are not 32-bit clean, limit event size to 2GB. K.O.
   const DWORD data_size = pevent->data_size; // 32-bit unsigned value
 
   if (data_size == 0) {
      cm_msg(MERROR, "bm_send_event", "invalid event data size zero");
      return BM_INVALID_SIZE;
   }
 
   if (data_size > MAX_DATA_SIZE) {
      cm_msg(MERROR, "bm_send_event", "invalid event data size %d (0x%x) maximum is %d (0x%x)", data_size, data_size, MAX_DATA_SIZE, MAX_DATA_SIZE);
      return BM_INVALID_SIZE;
   }
 
   const size_t event_size = sizeof(EVENT_HEADER) + data_size;
 
   size_t count = 0;
   for (int i=0; i<sg_n; i++) {
      count += sg_len[i];
   }
 
   if (count != event_size) {
      cm_msg(MERROR, "bm_send_event", "data size mismatch: event data_size %d, event_size %d not same as sum of sg_len %d", (int)data_size, (int)event_size, (int)count);
      return BM_INVALID_SIZE;
   }
 
   //printf("bm_send_event_sg: pevent %p, event_id 0x%04x, serial 0x%08x, data_size %d, event_size %d, total_size %d\n", pevent, pevent->event_id, pevent->serial_number, (int)pevent->data_size, (int)event_size, (int)total_size);
 
#ifdef LOCAL_ROUTINES
   {
      int status = 0;
      const size_t total_size = ALIGN8(event_size);
 
      BUFFER *pbuf = bm_get_buffer("bm_send_event_sg", buffer_handle, &status);
 
      if (!pbuf)
         return status;
 
      /* round up total_size to next DWORD boundary */
      //int total_size = ALIGN8(event_size);
 
      /* check if write cache is enabled */
      if (pbuf->write_cache_size) {
         status = bm_lock_buffer_write_cache(pbuf);
 
         if (status != BM_SUCCESS)
            return status;
         
         /* check if write cache is enabled */
         if (pbuf->write_cache_size) {
            size_t max_event_size = pbuf->write_cache_size/MAX_WRITE_CACHE_EVENT_SIZE_DIV;
            bool too_big = event_size > max_event_size;
 
            //printf("bm_send_event: write %zu/%zu max %zu, cache size %zu, wp %zu\n", event_size, total_size, max_event_size, pbuf->write_cache_size.load(), pbuf->write_cache_wp);
 
            /* if this event does not fit into the write cache, flush the write cache */
            if (pbuf->write_cache_wp > 0 && (pbuf->write_cache_wp + total_size > pbuf->write_cache_size || too_big)) {
               //printf("bm_send_event: write %zu/%zu but cache is full, size %zu, wp %zu\n", event_size, total_size, pbuf->write_cache_size.load(), pbuf->write_cache_wp);
 
               bm_lock_buffer_guard pbuf_guard(pbuf);
 
               if (!pbuf_guard.is_locked()) {
                  pbuf->write_cache_mutex.unlock();
                  return pbuf_guard.get_status();
               }
 
               int status = bm_flush_cache_locked(pbuf_guard, timeout_msec);
 
               if (pbuf_guard.is_locked()) {
                  // check if bm_wait_for_free_space() failed to relock the buffer
                  pbuf_guard.unlock();
               }
 
               if (status != BM_SUCCESS) {
                  pbuf->write_cache_mutex.unlock();
                  // bm_flush_cache() failed: timeout in bm_wait_for_free_space() or write cache size is bigger than buffer size or buffer was closed.
                  if (status == BM_NO_MEMORY)
                     cm_msg(MERROR, "bm_send_event", "write cache size is bigger than buffer size");
                  return status;
               }
 
               // write cache must be empty here
               assert(pbuf->write_cache_wp == 0);
            }
 
            /* write this event into the write cache, if it is not too big and if it fits */
            if (!too_big && pbuf->write_cache_wp + total_size <= pbuf->write_cache_size) {
               //printf("bm_send_event: write %d/%d to cache size %d, wp %d\n", (int)event_size, (int)total_size, (int)pbuf->write_cache_size, (int)pbuf->write_cache_wp);
               
               char* wptr = pbuf->write_cache + pbuf->write_cache_wp;
               
               for (int i=0; i<sg_n; i++) {
                  memcpy(wptr, sg_ptr[i], sg_len[i]);
                  wptr += sg_len[i];
               }
 
               pbuf->write_cache_wp += total_size;
 
               pbuf->write_cache_mutex.unlock();
               return BM_SUCCESS;
            }
         }
 
         /* event did not fit into the write cache, we flushed the write cache and we send it directly to shared memory */
         pbuf->write_cache_mutex.unlock();
      }
 
      /* we come here only for events that are too big to fit into the cache */
 
      /* lock the buffer */
      bm_lock_buffer_guard pbuf_guard(pbuf);
 
      if (!pbuf_guard.is_locked()) {
         return pbuf_guard.get_status();
      }
 
      /* calculate some shorthands */
      BUFFER_HEADER *pheader = pbuf->buffer_header;
 
#if 0
      status = bm_validate_buffer_locked(pbuf);
      if (status != BM_SUCCESS) {
         printf("bm_send_event: corrupted 111!\n");
         abort();
      }
#endif
 
      /* check if buffer is large enough */
      if (total_size >= (size_t)pheader->size) {
         pbuf_guard.unlock(); // unlock before cm_msg()
         cm_msg(MERROR, "bm_send_event", "total event size (%d) larger than size (%d) of buffer \'%s\'", (int)total_size, pheader->size, pheader->name);
         return BM_NO_MEMORY;
      }
 
      status = bm_wait_for_free_space_locked(pbuf_guard, timeout_msec, total_size, false);
 
      if (status != BM_SUCCESS) {
         // implicit unlock
         return status;
      }
 
#if 0
      status = bm_validate_buffer_locked(pbuf);
      if (status != BM_SUCCESS) {
         printf("bm_send_event: corrupted 222!\n");
         abort();
      }
#endif
 
      int old_write_pointer = pheader->write_pointer;
 
      bm_write_to_buffer_locked(pheader, sg_n, sg_ptr, sg_len, total_size);
 
      /* write pointer was incremented, but there should
       * always be some free space in the buffer and the
       * write pointer should never cacth up to the read pointer:
       * the rest of the code gets confused this happens (buffer 100% full)
       * as it is write_pointer == read_pointer can be either
       * 100% full or 100% empty. My solution: never fill
       * the buffer to 100% */
      assert(pheader->write_pointer != pheader->read_pointer);
 
      /* send wake up messages to all clients that want this event */
      int i;
      for (i = 0; i < pheader->max_client_index; i++) {
         BUFFER_CLIENT *pc = pheader->client + i;
         int request_id = bm_find_first_request_locked(pc, pevent);
         bm_notify_reader_locked(pheader, pc, old_write_pointer, request_id);
      }
 
#if 0
      status = bm_validate_buffer_locked(pbuf);
      if (status != BM_SUCCESS) {
         printf("bm_send_event: corrupted 333!\n");
         abort();
      }
#endif
 
      /* update statistics */
      pheader->num_in_events++;
      pbuf->count_sent += 1;
      pbuf->bytes_sent += total_size;
   }
#endif                          /* LOCAL_ROUTINES */
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_send_event_vec() [1/2]

int bm_send_event_vec	(	int	buffer_handle,
		const std::vector< char > &	event,
		int	timeout_msec
	)

Definition at line 9713 of file midas.cxx.

{
   const char* cptr = event.data();
   size_t clen = event.size();
   return bm_send_event_sg(buffer_handle, 1, &cptr, &clen, timeout_msec);
}

Here is the call graph for this function:

bm_send_event_vec() [2/2]

int bm_send_event_vec	(	int	buffer_handle,
		const std::vector< std::vector< char > > &	event,
		int	timeout_msec
	)

Definition at line 9720 of file midas.cxx.

{
   int sg_n = event.size();
   const char* sg_ptr[sg_n];
   size_t sg_len[sg_n];
   for (int i=0; i<sg_n; i++) {
      sg_ptr[i] = event[i].data();
      sg_len[i] = event[i].size();
   }
   return bm_send_event_sg(buffer_handle, sg_n, sg_ptr, sg_len, timeout_msec);
}

Here is the call graph for this function:

bm_set_cache_size()

INT bm_set_cache_size	(	INT	buffer_handle,
		size_t	read_size,
		size_t	write_size
	)

Modifies buffer cache size. Without a buffer cache, events are copied to/from the shared memory event by event.

To protect processed from accessing the shared memory simultaneously, semaphores are used. Since semaphore operations are CPU consuming (typically 50-100us) this can slow down the data transfer especially for small events. By using a cache the number of semaphore operations is reduced dramatically. Instead writing directly to the shared memory, the events are copied to a local cache buffer. When this buffer is full, it is copied to the shared memory in one operation. The same technique can be used when receiving events.

The drawback of this method is that the events have to be copied twice, once to the cache and once from the cache to the shared memory. Therefore it can happen that the usage of a cache even slows down data throughput on a given environment (computer type, OS type, event size). The cache size has therefore be optimized manually to maximize data throughput.

Parameters

buffer_handle	buffer handle obtained via bm_open_buffer()
read_size	cache size for reading events in bytes, zero for no cache
write_size	cache size for writing events in bytes, zero for no cache

Returns

BM_SUCCESS, BM_INVALID_HANDLE, BM_NO_MEMORY, BM_INVALID_PARAM

Definition at line 8148 of file midas.cxx.

{
   if (rpc_is_remote())
      return rpc_call(RPC_BM_SET_CACHE_SIZE, buffer_handle, read_size, write_size);
 
#ifdef LOCAL_ROUTINES
   {
      int status = 0;
 
      BUFFER *pbuf = bm_get_buffer("bm_set_cache_size", buffer_handle, &status);
 
      if (!pbuf)
         return status;
 
      /* lock pbuf for local access. we do not lock buffer semaphore because we do not touch the shared memory */
 
      status = bm_lock_buffer_mutex(pbuf);
 
      if (status != BM_SUCCESS)
         return status;
 
      if (write_size < 0)
         write_size = 0;
 
      if (write_size > 0) {
         if (write_size < MIN_WRITE_CACHE_SIZE) {
            cm_msg(MERROR, "bm_set_cache_size", "requested write cache size %zu on buffer \"%s\" too small, will use minimum size %d", write_size, pbuf->buffer_name, MIN_WRITE_CACHE_SIZE);
            write_size = MIN_WRITE_CACHE_SIZE;
         }
      }
 
      size_t max_write_size = pbuf->buffer_header->size/MAX_WRITE_CACHE_SIZE_DIV;
 
      if (write_size > max_write_size) {
         size_t new_write_size = max_write_size;
         cm_msg(MERROR, "bm_set_cache_size", "requested write cache size %zu on buffer \"%s\" is too big: buffer size is %d, write cache size will be %zu bytes", write_size, pbuf->buffer_name, pbuf->buffer_header->size, new_write_size);
         write_size = new_write_size;
      }
 
      pbuf->buffer_mutex.unlock();
 
      /* resize read cache */
 
      status = bm_lock_buffer_read_cache(pbuf);
 
      if (status != BM_SUCCESS) {
         return status;
      }
 
      if (pbuf->read_cache_size > 0) {
         free(pbuf->read_cache);
         pbuf->read_cache = NULL;
      }
 
      if (read_size > 0) {
         pbuf->read_cache = (char *) malloc(read_size);
         if (pbuf->read_cache == NULL) {
            pbuf->read_cache_size = 0;
            pbuf->read_cache_rp = 0;
            pbuf->read_cache_wp = 0;
            pbuf->read_cache_mutex.unlock();
            cm_msg(MERROR, "bm_set_cache_size", "not enough memory to allocate read cache for buffer \"%s\", malloc(%zu) failed", pbuf->buffer_name, read_size);
            return BM_NO_MEMORY;
         }
      }
 
      pbuf->read_cache_size = read_size;
      pbuf->read_cache_rp = 0;
      pbuf->read_cache_wp = 0;
 
      pbuf->read_cache_mutex.unlock();
 
      /* resize the write cache */
 
      status = bm_lock_buffer_write_cache(pbuf);
 
      if (status != BM_SUCCESS)
         return status;
 
      // FIXME: should flush the write cache!
      if (pbuf->write_cache_size && pbuf->write_cache_wp > 0) {
         cm_msg(MERROR, "bm_set_cache_size", "buffer \"%s\" lost %zu bytes from the write cache", pbuf->buffer_name, pbuf->write_cache_wp);
      }
 
      /* manage write cache */
      if (pbuf->write_cache_size > 0) {
         free(pbuf->write_cache);
         pbuf->write_cache = NULL;
      }
 
      if (write_size > 0) {
         pbuf->write_cache = (char *) M_MALLOC(write_size);
         if (pbuf->write_cache == NULL) {
            pbuf->write_cache_size = 0;
            pbuf->write_cache_rp = 0;
            pbuf->write_cache_wp = 0;
            pbuf->write_cache_mutex.unlock();
            cm_msg(MERROR, "bm_set_cache_size", "not enough memory to allocate write cache for buffer \"%s\", malloc(%zu) failed", pbuf->buffer_name, write_size);
            return BM_NO_MEMORY;
         }
      }
 
      pbuf->write_cache_size = write_size;
      pbuf->write_cache_rp = 0;
      pbuf->write_cache_wp = 0;
 
      pbuf->write_cache_mutex.unlock();
   }
#endif                          /* LOCAL_ROUTINES */
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_skip_event() [1/2]

static int bm_skip_event ( BUFFER *  pbuf )

static

Definition at line 10841 of file midas.cxx.

{
   /* clear read cache */
   if (pbuf->read_cache_size > 0) {
 
      int status = bm_lock_buffer_read_cache(pbuf);
 
      if (status != BM_SUCCESS)
         return status;
 
      pbuf->read_cache_rp = 0;
      pbuf->read_cache_wp = 0;
 
      pbuf->read_cache_mutex.unlock();
   }
   
   bm_lock_buffer_guard pbuf_guard(pbuf);
 
   if (!pbuf_guard.is_locked())
      return pbuf_guard.get_status();
   
   BUFFER_HEADER *pheader = pbuf->buffer_header;
   
   /* forward read pointer to global write pointer */
   BUFFER_CLIENT *pclient = bm_get_my_client_locked(pbuf_guard);
   pclient->read_pointer = pheader->write_pointer;
   
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_skip_event() [2/2]

INT bm_skip_event

(

INT

buffer_handle

)

Skip all events in current buffer.

Useful for single event displays to see the newest events

Parameters

buffer_handle

Handle of the buffer. Must be obtained via bm_open_buffer.

Returns

BM_SUCCESS, BM_INVALID_HANDLE, RPC_NET_ERROR

Definition at line 10882 of file midas.cxx.

                                     {
   if (rpc_is_remote())
      return rpc_call(RPC_BM_SKIP_EVENT, buffer_handle);
 
#ifdef LOCAL_ROUTINES
   {
      int status = 0;
 
      BUFFER *pbuf = bm_get_buffer("bm_skip_event", buffer_handle, &status);
 
      if (!pbuf)
         return status;
 
      return bm_skip_event(pbuf);
   }
#endif
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

bm_update_last_activity()

static void bm_update_last_activity ( DWORD  millitime )

static

Update last activity time

Definition at line 6125 of file midas.cxx.

                                                     {
   int pid = ss_getpid();
 
   std::vector<BUFFER*> mybuffers;
 
   gBuffersMutex.lock();
   mybuffers = gBuffers;
   gBuffersMutex.unlock();
 
   for (BUFFER* pbuf : mybuffers) {
      if (!pbuf)
         continue;
      if (pbuf->attached) {
 
         bm_lock_buffer_guard pbuf_guard(pbuf);
 
         if (!pbuf_guard.is_locked())
            continue;
 
         BUFFER_HEADER *pheader = pbuf->buffer_header;
         for (int j = 0; j < pheader->max_client_index; j++) {
            BUFFER_CLIENT *pclient = pheader->client + j;
            if (pclient->pid == pid) {
               pclient->last_activity = millitime;
            }
         }
      }
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_update_read_pointer_locked()

static BOOL bm_update_read_pointer_locked ( const char *  caller_name, BUFFER_HEADER *  pheader  )

static

Definition at line 8728 of file midas.cxx.

                                                                                           {
   assert(caller_name);
 
   /* calculate global read pointer as "minimum" of client read pointers */
   int min_rp = pheader->write_pointer;
 
   int i;
   for (i = 0; i < pheader->max_client_index; i++) {
      BUFFER_CLIENT *pc = pheader->client + i;
      if (pc->pid) {
         bm_validate_client_pointers_locked(pheader, pc);
 
#if 0
         printf("bm_update_read_pointer: [%s] rp %d, wp %d, size %d, min_rp %d, client [%s] rp %d\n",
                pheader->name,
                pheader->read_pointer,
                pheader->write_pointer,
                pheader->size,
                min_rp,
                pc->name,
                pc->read_pointer);
#endif
 
         if (pheader->read_pointer <= pheader->write_pointer) {
            // normal pointers
            if (pc->read_pointer < min_rp)
               min_rp = pc->read_pointer;
         } else {
            // inverted pointers
            if (pc->read_pointer <= pheader->write_pointer) {
               // clients 3 and 4
               if (pc->read_pointer < min_rp)
                  min_rp = pc->read_pointer;
            } else {
               // clients 1 and 2
               int xptr = pc->read_pointer - pheader->size;
               if (xptr < min_rp)
                  min_rp = xptr;
            }
         }
      }
   }
 
   if (min_rp < 0)
      min_rp += pheader->size;
 
   assert(min_rp >= 0);
   assert(min_rp < pheader->size);
 
   if (min_rp == pheader->read_pointer) {
      return FALSE;
   }
 
#if 0
   printf("bm_update_read_pointer: [%s] rp %d, wp %d, size %d, new_rp %d, moved\n",
          pheader->name,
          pheader->read_pointer,
          pheader->write_pointer,
          pheader->size,
          min_rp);
#endif
 
   pheader->read_pointer = min_rp;
 
   return TRUE;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_validate_buffer_locked()

static int bm_validate_buffer_locked ( const BUFFER *  pbuf )

static

Definition at line 6315 of file midas.cxx.

                                                         {
   const BUFFER_HEADER *pheader = pbuf->buffer_header;
   const char *pdata = (const char *) (pheader + 1);
 
   //printf("bm_validate_buffer: buffer \"%s\"\n", pheader->name);
 
   //printf("size: %d, rp: %d, wp: %d\n", pheader->size, pheader->read_pointer, pheader->write_pointer);
 
   //printf("clients: max: %d, num: %d, MAX_CLIENTS: %d\n", pheader->max_client_index, pheader->num_clients, MAX_CLIENTS);
 
   if (pheader->read_pointer < 0 || pheader->read_pointer >= pheader->size) {
      cm_msg(MERROR, "bm_validate_buffer",
             "buffer \"%s\" is corrupted: invalid read pointer %d. Size %d, write pointer %d", pheader->name,
             pheader->read_pointer, pheader->size, pheader->write_pointer);
      return BM_CORRUPTED;
   }
 
   if (pheader->write_pointer < 0 || pheader->write_pointer >= pheader->size) {
      cm_msg(MERROR, "bm_validate_buffer",
             "buffer \"%s\" is corrupted: invalid write pointer %d. Size %d, read pointer %d", pheader->name,
             pheader->write_pointer, pheader->size, pheader->read_pointer);
      return BM_CORRUPTED;
   }
 
   if (!bm_validate_rp("bm_validate_buffer_locked", pheader, pheader->read_pointer)) {
      cm_msg(MERROR, "bm_validate_buffer", "buffer \"%s\" is corrupted: read pointer %d is invalid", pheader->name,
             pheader->read_pointer);
      return BM_CORRUPTED;
   }
 
   int rp = pheader->read_pointer;
   int rp0 = -1;
   while (rp != pheader->write_pointer) {
      if (!bm_validate_rp("bm_validate_buffer_locked", pheader, rp)) {
         cm_msg(MERROR, "bm_validate_buffer", "buffer \"%s\" is corrupted: invalid rp %d, last good event at rp %d",
                pheader->name, rp, rp0);
         return BM_CORRUPTED;
      }
      //bm_print_event(pdata, rp);
      int rp1 = bm_next_rp("bm_validate_buffer_locked", pheader, pdata, rp);
      if (rp1 < 0) {
         cm_msg(MERROR, "bm_validate_buffer",
                "buffer \"%s\" is corrupted: invalid event at rp %d, last good event at rp %d", pheader->name, rp, rp0);
         return BM_CORRUPTED;
      }
      rp0 = rp;
      rp = rp1;
   }
 
   int i;
   for (i = 0; i < MAX_CLIENTS; i++) {
      const BUFFER_CLIENT *c = &pheader->client[i];
      if (c->pid == 0)
         continue;
      BOOL get_all = FALSE;
      int j;
      for (j = 0; j < MAX_EVENT_REQUESTS; j++) {
         const EVENT_REQUEST *r = &c->event_request[j];
         if (!r->valid)
            continue;
         BOOL xget_all = r->sampling_type == GET_ALL;
         get_all = (get_all || xget_all);
         //printf("client slot %d: pid %d, name \"%s\", request %d: id %d, valid %d, sampling_type %d, get_all %d\n", i, c->pid, c->name, j, r->id, r->valid, r->sampling_type, xget_all);
      }
 
      int rp = c->read_pointer;
      int rp0 = -1;
      while (rp != pheader->write_pointer) {
         //bm_print_event(pdata, rp);
         int rp1 = bm_next_rp("bm_validate_buffer_locked", pheader, pdata, rp);
         if (rp1 < 0) {
            cm_msg(MERROR, "bm_validate_buffer",
                   "buffer \"%s\" is corrupted for client \"%s\" rp %d: invalid event at rp %d, last good event at rp %d",
                   pheader->name, c->name, c->read_pointer, rp, rp0);
            return BM_CORRUPTED;
         }
         rp0 = rp;
         rp = rp1;
      }
   }
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_validate_client_index_locked()

static int bm_validate_client_index_locked ( bm_lock_buffer_guard &  pbuf_guard )

static

Definition at line 5930 of file midas.cxx.

{
   const BUFFER *pbuf = pbuf_guard.get_pbuf();
 
   bool badindex  = false;
   bool badclient = false;
 
   int idx = pbuf->client_index;
 
   if (idx < 0) {
      badindex = true;
   } else if (idx > pbuf->buffer_header->max_client_index) {
      badindex = true;
   } else {
      BUFFER_CLIENT *pclient = &pbuf->buffer_header->client[idx];
      if (pclient->name[0] == 0)
         badclient = true;
      else if (pclient->pid != ss_getpid())
         badclient = true;
 
      //if (strcmp(pclient->name,"mdump")==0) {
      //   for (int i=0; i<15; i++) {
      //      printf("sleep %d\n", i);
      //      ::sleep(1);
      //   }
      //}
   }
 
#if 0
   if (badindex) {
      printf("bm_validate_client_index: pbuf=%p, buf_name \"%s\", client_index=%d, max_client_index=%d, badindex %d, pid=%d\n",
             pbuf, pbuf->buffer_header->name, pbuf->client_index, pbuf->buffer_header->max_client_index,
             badindex, ss_getpid());
   } else if (badclient) {
      printf("bm_validate_client_index: pbuf=%p, buf_name \"%s\", client_index=%d, max_client_index=%d, client_name=\'%s\', client_pid=%d, pid=%d, badclient %d\n",
             pbuf, pbuf->buffer_header->name, pbuf->client_index, pbuf->buffer_header->max_client_index,
             pbuf->buffer_header->client[idx].name, pbuf->buffer_header->client[idx].pid,
             ss_getpid(), badclient);
   } else {
      printf("bm_validate_client_index: pbuf=%p, buf_name \"%s\", client_index=%d, max_client_index=%d, client_name=\'%s\', client_pid=%d, pid=%d, goodclient\n",
             pbuf, pbuf->buffer_header->name, pbuf->client_index, pbuf->buffer_header->max_client_index,
             pbuf->buffer_header->client[idx].name, pbuf->buffer_header->client[idx].pid,
             ss_getpid());
   }
#endif
 
   if (badindex || badclient) {
      static int prevent_recursion = 1;
 
      if (prevent_recursion) {
         prevent_recursion = 0;
 
         if (badindex) {
            cm_msg(MERROR, "bm_validate_client_index", "My client index %d in buffer \'%s\' is invalid, max_client_index %d, my pid %d", idx, pbuf->buffer_header->name, pbuf->buffer_header->max_client_index, ss_getpid());
         } else {
            cm_msg(MERROR, "bm_validate_client_index", "My client index %d in buffer \'%s\' is invalid: client name \'%s\', pid %d should be my pid %d", idx, pbuf->buffer_header->name, pbuf->buffer_header->client[idx].name, pbuf->buffer_header->client[idx].pid, ss_getpid());
         }
 
         cm_msg(MERROR, "bm_validate_client_index", "Maybe this client was removed by a timeout. See midas.log. Cannot continue, aborting...");
      }
 
      if (badindex) {
         fprintf(stderr, "bm_validate_client_index: My client index %d in buffer \'%s\' is invalid, max_client_index %d, my pid %d\n", idx, pbuf->buffer_header->name, pbuf->buffer_header->max_client_index, ss_getpid());
      } else {
         fprintf(stderr, "bm_validate_client_index: My client index %d in buffer \'%s\' is invalid: client name \'%s\', pid %d should be my pid %d\n", idx, pbuf->buffer_header->name, pbuf->buffer_header->client[idx].name, pbuf->buffer_header->client[idx].pid, ss_getpid());
      }
      
      fprintf(stderr, "bm_validate_client_index: Maybe this client was removed by a timeout. See midas.log. Cannot continue, aborting...\n");
 
      pbuf_guard.unlock();
 
      abort();
   }
 
   return idx;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_validate_client_pointers_locked()

static void bm_validate_client_pointers_locked ( const BUFFER_HEADER *  pheader, BUFFER_CLIENT *  pclient  )

static

Definition at line 8630 of file midas.cxx.

                                                                                                     {
   assert(pheader->read_pointer >= 0 && pheader->read_pointer <= pheader->size);
   assert(pclient->read_pointer >= 0 && pclient->read_pointer <= pheader->size);
 
   if (pheader->read_pointer <= pheader->write_pointer) {
 
      if (pclient->read_pointer < pheader->read_pointer) {
         cm_msg(MINFO, "bm_validate_client_pointers",
                "Corrected read pointer for client \'%s\' on buffer \'%s\' from %d to %d, write pointer %d, size %d",
                pclient->name,
                pheader->name, pclient->read_pointer, pheader->read_pointer, pheader->write_pointer, pheader->size);
 
         pclient->read_pointer = pheader->read_pointer;
      }
 
      if (pclient->read_pointer > pheader->write_pointer) {
         cm_msg(MINFO, "bm_validate_client_pointers",
                "Corrected read pointer for client \'%s\' on buffer \'%s\' from %d to %d, read pointer %d, size %d",
                pclient->name,
                pheader->name, pclient->read_pointer, pheader->write_pointer, pheader->read_pointer, pheader->size);
 
         pclient->read_pointer = pheader->write_pointer;
      }
 
   } else {
 
      if (pclient->read_pointer < 0) {
         cm_msg(MINFO, "bm_validate_client_pointers",
                "Corrected read pointer for client \'%s\' on buffer \'%s\' from %d to %d, write pointer %d, size %d",
                pclient->name,
                pheader->name, pclient->read_pointer, pheader->read_pointer, pheader->write_pointer, pheader->size);
 
         pclient->read_pointer = pheader->read_pointer;
      }
 
      if (pclient->read_pointer >= pheader->size) {
         cm_msg(MINFO, "bm_validate_client_pointers",
                "Corrected read pointer for client \'%s\' on buffer \'%s\' from %d to %d, write pointer %d, size %d",
                pclient->name,
                pheader->name, pclient->read_pointer, pheader->read_pointer, pheader->write_pointer, pheader->size);
 
         pclient->read_pointer = pheader->read_pointer;
      }
 
      if (pclient->read_pointer > pheader->write_pointer && pclient->read_pointer < pheader->read_pointer) {
         cm_msg(MINFO, "bm_validate_client_pointers",
                "Corrected read pointer for client \'%s\' on buffer \'%s\' from %d to %d, write pointer %d, size %d",
                pclient->name,
                pheader->name, pclient->read_pointer, pheader->read_pointer, pheader->write_pointer, pheader->size);
 
         pclient->read_pointer = pheader->read_pointer;
      }
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_validate_rp()

static BOOL bm_validate_rp ( const char *  who, const BUFFER_HEADER *  pheader, int  rp  )

static

Definition at line 6197 of file midas.cxx.

                                                                                  {
   if (rp < 0 || rp > pheader->size) {
      cm_msg(MERROR, "bm_validate_rp",
             "error: buffer \"%s\" is corrupted: rp %d is invalid. buffer read_pointer %d, write_pointer %d, size %d, called from %s",
             pheader->name,
             rp,
             pheader->read_pointer,
             pheader->write_pointer,
             pheader->size,
             who);
      return FALSE;
   }
 
   if ((rp + (int) sizeof(EVENT_HEADER)) > pheader->size) {
      // note ">" here, has to match bm_incr_rp() and bm_write_to_buffer()
      cm_msg(MERROR, "bm_validate_rp",
             "error: buffer \"%s\" is corrupted: rp %d plus event header point beyond the end of buffer by %d bytes. buffer read_pointer %d, write_pointer %d, size %d, called from %s",
             pheader->name,
             rp,
             (int) (rp + sizeof(EVENT_HEADER) - pheader->size),
             pheader->read_pointer,
             pheader->write_pointer,
             pheader->size,
             who);
      return FALSE;
   }
 
   return TRUE;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_wait_for_free_space_locked()

static int bm_wait_for_free_space_locked ( bm_lock_buffer_guard &  pbuf_guard, int  timeout_msec, int  requested_space, bool  unlock_write_cache  )

static

• signal other clients wait mode *‍/

• validate client index: we could have been removed from the buffer *‍/

Definition at line 9088 of file midas.cxx.

{
   // return values:
   // BM_SUCCESS - have "requested_space" bytes free in the buffer
   // BM_CORRUPTED - shared memory is corrupted
   // BM_NO_MEMORY - asked for more than buffer size
   // BM_ASYNC_RETURN - timeout waiting for free space
   // BM_INVALID_HANDLE - buffer was closed (locks released) (via bm_clock_xxx())
   // SS_ABORT - we are told to shutdown (locks releases)
 
   int status;
   BUFFER* pbuf = pbuf_guard.get_pbuf();
   BUFFER_HEADER *pheader = pbuf->buffer_header;
   char *pdata = (char *) (pheader + 1);
 
   /* make sure the buffer never completely full:
    * read pointer and write pointer would coincide
    * and the code cannot tell if it means the
    * buffer is 100% full or 100% empty. It will explode
    * or lose events */
   requested_space += 100;
 
   if (requested_space >= pheader->size)
      return BM_NO_MEMORY;
 
   DWORD time_start = ss_millitime();
   DWORD time_end = time_start + timeout_msec;
 
   //DWORD blocking_time = 0;
   //int blocking_loops = 0;
   int blocking_client_index = -1;
   char blocking_client_name[NAME_LENGTH];
   blocking_client_name[0] = 0;
 
   while (1) {
      while (1) {
         /* check if enough space in buffer */
 
         int free = pheader->read_pointer - pheader->write_pointer;
         if (free <= 0)
            free += pheader->size;
 
         //printf("bm_wait_for_free_space: buffer pointers: read: %d, write: %d, free space: %d, bufsize: %d, event size: %d, timeout %d\n", pheader->read_pointer, pheader->write_pointer, free, pheader->size, requested_space, timeout_msec);
 
         if (requested_space < free) { /* note the '<' to avoid 100% filling */
            //if (blocking_loops) {
            //   DWORD wait_time = ss_millitime() - blocking_time;
            //   printf("blocking client \"%s\", time %d ms, loops %d\n", blocking_client_name, wait_time, blocking_loops);
            //}
 
            if (pbuf->wait_start_time != 0) {
               DWORD now = ss_millitime();
               DWORD wait_time = now - pbuf->wait_start_time;
               pbuf->time_write_wait += wait_time;
               pbuf->wait_start_time = 0;
               int iclient = pbuf->wait_client_index;
               //printf("bm_wait_for_free_space: wait ended: wait time %d ms, blocking client index %d\n", wait_time, iclient);
               if (iclient >= 0 && iclient < MAX_CLIENTS) {
                  pbuf->client_count_write_wait[iclient] += 1;
                  pbuf->client_time_write_wait[iclient] += wait_time;
               }
            }
 
            //if (blocking_loops > 0) {
            //   printf("bm_wait_for_free_space: buffer pointers: read: %d, write: %d, free space: %d, bufsize: %d, event size: %d, timeout %d, found space after %d waits\n", pheader->read_pointer, pheader->write_pointer, free, pheader->size, requested_space, timeout_msec, blocking_loops);
            //}
 
            return BM_SUCCESS;
         }
 
         if (!bm_validate_rp("bm_wait_for_free_space_locked", pheader, pheader->read_pointer)) {
            cm_msg(MERROR, "bm_wait_for_free_space",
                   "error: buffer \"%s\" is corrupted: read_pointer %d, write_pointer %d, size %d, free %d, waiting for %d bytes: read pointer is invalid",
                   pheader->name,
                   pheader->read_pointer,
                   pheader->write_pointer,
                   pheader->size,
                   free,
                   requested_space);
            return BM_CORRUPTED;
         }
 
         const EVENT_HEADER *pevent = (const EVENT_HEADER *) (pdata + pheader->read_pointer);
         int event_size = pevent->data_size + sizeof(EVENT_HEADER);
         int total_size = ALIGN8(event_size);
 
#if 0
         printf("bm_wait_for_free_space: buffer pointers: read: %d, write: %d, free space: %d, bufsize: %d, event size: %d, blocking event size %d/%d\n", pheader->read_pointer, pheader->write_pointer, free, pheader->size, requested_space, event_size, total_size);
#endif
 
         if (pevent->data_size <= 0 || total_size <= 0 || total_size > pheader->size) {
            cm_msg(MERROR, "bm_wait_for_free_space",
                   "error: buffer \"%s\" is corrupted: read_pointer %d, write_pointer %d, size %d, free %d, waiting for %d bytes: read pointer points to an invalid event: data_size %d, event size %d, total_size %d",
                   pheader->name,
                   pheader->read_pointer,
                   pheader->write_pointer,
                   pheader->size,
                   free,
                   requested_space,
                   pevent->data_size,
                   event_size,
                   total_size);
            return BM_CORRUPTED;
         }
 
         int blocking_client = -1;
 
         int i;
         for (i = 0; i < pheader->max_client_index; i++) {
            BUFFER_CLIENT *pc = pheader->client + i;
            if (pc->pid) {
               if (pc->read_pointer == pheader->read_pointer) {
                  /*
                    First assume that the client with the "minimum" read pointer
                    is not really blocking due to a GET_ALL request.
                  */
                  BOOL blocking = FALSE;
                  //int blocking_request_id = -1;
 
                  int j;
                  for (j = 0; j < pc->max_request_index; j++) {
                     const EVENT_REQUEST *prequest = pc->event_request + j;
                     if (prequest->valid
                         && bm_match_event(prequest->event_id, prequest->trigger_mask, pevent)) {
                        if (prequest->sampling_type & GET_ALL) {
                           blocking = TRUE;
                           //blocking_request_id = prequest->id;
                           break;
                        }
                     }
                  }
 
                  //printf("client [%s] blocking %d, request %d\n", pc->name, blocking, blocking_request_id);
 
                  if (blocking) {
                     blocking_client = i;
                     break;
                  }
 
                  pc->read_pointer = bm_incr_rp_no_check(pheader, pc->read_pointer, total_size);
               }
            }
         } /* client loop */
 
         if (blocking_client >= 0) {
            blocking_client_index = blocking_client;
            mstrlcpy(blocking_client_name, pheader->client[blocking_client].name, sizeof(blocking_client_name));
            //if (!blocking_time) {
            //   blocking_time = ss_millitime();
            //}
 
            //printf("bm_wait_for_free_space: buffer pointers: read: %d, write: %d, free space: %d, bufsize: %d, event size: %d, timeout %d, must wait for more space!\n", pheader->read_pointer, pheader->write_pointer, free, pheader->size, requested_space, timeout_msec);
 
            // from this "break" we go into timeout check and sleep/wait.
            break;
         }
 
         /* no blocking clients. move the read pointer and again check for free space */
 
         BOOL moved = bm_update_read_pointer_locked("bm_wait_for_free_space", pheader);
 
         if (!moved) {
            cm_msg(MERROR, "bm_wait_for_free_space",
                   "error: buffer \"%s\" is corrupted: read_pointer %d, write_pointer %d, size %d, free %d, waiting for %d bytes: read pointer did not move as expected",
                   pheader->name,
                   pheader->read_pointer,
                   pheader->write_pointer,
                   pheader->size,
                   free,
                   requested_space);
            return BM_CORRUPTED;
         }
 
         /* we freed one event, loop back to the check for free space */
      }
 
      //blocking_loops++;
 
      /* at least one client is blocking */
 
      BUFFER_CLIENT *pc = bm_get_my_client_locked(pbuf_guard);
      pc->write_wait = requested_space;
 
      if (pbuf->wait_start_time == 0) {
         pbuf->wait_start_time = ss_millitime();
         pbuf->count_write_wait++;
         if (requested_space > pbuf->max_requested_space)
            pbuf->max_requested_space = requested_space;
         pbuf->wait_client_index = blocking_client_index;
      }
 
      DWORD now = ss_millitime();
 
      //printf("bm_wait_for_free_space: start 0x%08x, now 0x%08x, end 0x%08x, timeout %d, wait %d\n", time_start, now, time_end, timeout_msec, time_end - now);
 
      int sleep_time_msec = 1000;
 
      if (timeout_msec == BM_WAIT) {
         // wait forever
      } else if (timeout_msec == BM_NO_WAIT) {
         // no wait
         return BM_ASYNC_RETURN;
      } else {
         // check timeout
         if (now >= time_end) {
            // timeout!
            return BM_ASYNC_RETURN;
         }
 
         sleep_time_msec = time_end - now;
 
         if (sleep_time_msec <= 0) {
            sleep_time_msec = 10;
         } else if (sleep_time_msec > 1000) {
            sleep_time_msec = 1000;
         }
      }
 
      ss_suspend_get_buffer_port(ss_gettid(), &pc->port);
 
      /* before waiting, unlock everything in the correct order */
 
      pbuf_guard.unlock();
 
      if (unlock_write_cache)
         pbuf->write_cache_mutex.unlock();
 
      //printf("bm_wait_for_free_space: blocking client \"%s\"\n", blocking_client_name);
 
#ifdef DEBUG_MSG
      cm_msg(MDEBUG, "Send sleep: rp=%d, wp=%d, level=%1.1lf", pheader->read_pointer, pheader->write_pointer, 100 - 100.0 * size / pheader->size);
#endif
 
      //int idx = bm_validate_client_index_locked(pbuf, FALSE);
      //if (idx >= 0)
      //   pheader->client[idx].write_wait = requested_space;
 
      //bm_cleanup("bm_wait_for_free_space", ss_millitime(), FALSE);
 
      status = ss_suspend(sleep_time_msec, MSG_BM);
 
      /* we are told to shutdown */
      if (status == SS_ABORT) {
         // NB: buffer is locked!
         return SS_ABORT;
      }
 
      /* make sure we do sleep in this loop:
       * if we are the mserver receiving data on the event
       * socket and the data buffer is full, ss_suspend() will
       * never sleep: it will detect data on the event channel,
       * call rpc_server_receive() (recursively, we already *are* in
       * rpc_server_receive()) and return without sleeping. Result
       * is a busy loop waiting for free space in data buffer */
 
      /* update May 2021: ss_suspend(MSG_BM) no longer looks at
       * the event socket, and should sleep now, so this sleep below
       * maybe is not needed now. but for safety, I keep it. K.O. */
 
      if (status != SS_TIMEOUT) {
         //printf("ss_suspend: status %d\n", status);
         ss_sleep(1);
      }
 
      /* we may be stuck in this loop for an arbitrary long time,
       * depending on how other buffer clients read the accumulated data
       * so we should update all the timeouts & etc. K.O. */
 
      cm_periodic_tasks();
 
      /* lock things again in the correct order */
 
      if (unlock_write_cache) {
         status = bm_lock_buffer_write_cache(pbuf);
 
         if (status != BM_SUCCESS) {
            // bail out with all locks released
            return status;
         }
      }
 
      if (!pbuf_guard.relock()) {
         if (unlock_write_cache) {
            pbuf->write_cache_mutex.unlock();
         }
 
         // bail out with all locks released
         return pbuf_guard.get_status();
      }
 
      /* revalidate the client index: we could have been removed from the buffer while sleeping */
      pc = bm_get_my_client_locked(pbuf_guard);
 
      pc->write_wait = 0;
 
      //idx = bm_validate_client_index_locked(pbuf, FALSE);
      //if (idx >= 0)
      //   pheader->client[idx].write_wait = 0;
      //else {
      //   cm_msg(MERROR, "bm_wait_for_free_space", "our client index is no longer valid, exiting...");
      //   status = SS_ABORT;
      //}
 
#ifdef DEBUG_MSG
      cm_msg(MDEBUG, "Send woke up: rp=%d, wp=%d, level=%1.1lf", pheader->read_pointer, pheader->write_pointer, 100 - 100.0 * size / pheader->size);
#endif
 
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_wait_for_more_events_locked()

static int bm_wait_for_more_events_locked ( bm_lock_buffer_guard &  pbuf_guard, BUFFER_CLIENT *  pc, int  timeout_msec, BOOL  unlock_read_cache  )

static

Definition at line 9400 of file midas.cxx.

{
   BUFFER* pbuf = pbuf_guard.get_pbuf();
   BUFFER_HEADER* pheader = pbuf->buffer_header;
   
   //printf("bm_wait_for_more_events_locked: [%s] timeout %d\n", pheader->name, timeout_msec);
   
   if (pc->read_pointer != pheader->write_pointer) {
      // buffer has data
      return BM_SUCCESS;
   }
 
   if (timeout_msec == BM_NO_WAIT) {
      /* event buffer is empty and we are told to not wait */
      if (!pc->read_wait) {
         //printf("bm_wait_for_more_events: buffer [%s] client [%s] set read_wait in BM_NO_WAIT!\n", pheader->name, pc->name);
         pc->read_wait = TRUE;
      }
      return BM_ASYNC_RETURN;
   }
 
   DWORD time_start = ss_millitime();
   DWORD time_wait  = time_start + timeout_msec;
   DWORD sleep_time = 1000;
   if (timeout_msec == BM_NO_WAIT) {
      // default sleep time
   } else if (timeout_msec == BM_WAIT) {
      // default sleep time
   } else {
      if (sleep_time > (DWORD)timeout_msec)
         sleep_time = timeout_msec;
   }
 
   //printf("time start 0x%08x, end 0x%08x, sleep %d\n", time_start, time_wait, sleep_time);
 
   while (pc->read_pointer == pheader->write_pointer) {
      /* wait until there is data in the buffer (write pointer moves) */
 
      if (!pc->read_wait) {
         //printf("bm_wait_for_more_events: buffer [%s] client [%s] set read_wait!\n", pheader->name, pc->name);
         pc->read_wait = TRUE;
      }
 
      pc->last_activity = ss_millitime();
 
      ss_suspend_get_buffer_port(ss_gettid(), &pc->port);
 
      // NB: locking order is: 1st read cache lock, 2nd buffer lock, unlock in reverse order
 
      pbuf_guard.unlock();
 
      if (unlock_read_cache)
         pbuf->read_cache_mutex.unlock();
 
      int status = ss_suspend(sleep_time, MSG_BM);
 
      if (timeout_msec == BM_NO_WAIT) {
         // return immediately
      } else if (timeout_msec == BM_WAIT) {
         // wait forever
      } else {
         DWORD now = ss_millitime();
         //printf("check timeout: now 0x%08x, end 0x%08x, diff %d\n", now, time_wait, time_wait - now);
         if (now >= time_wait) {
            timeout_msec = BM_NO_WAIT; // cause immediate return
         } else {
            sleep_time = time_wait - now;
            if (sleep_time > 1000)
               sleep_time = 1000;
            //printf("time start 0x%08x, now 0x%08x, end 0x%08x, sleep %d\n", time_start, now, time_wait, sleep_time);
         }
      }
 
      // NB: locking order is: 1st read cache lock, 2nd buffer lock, unlock in reverse order
 
      if (unlock_read_cache) {
         status = bm_lock_buffer_read_cache(pbuf);
         if (status != BM_SUCCESS) {
            // bail out with all locks released
            return status;
         }
      }
 
      if (!pbuf_guard.relock()) {
         if (unlock_read_cache) {
            pbuf->read_cache_mutex.unlock();
         }
         // bail out with all locks released
         return pbuf_guard.get_status();
      }
      
      /* need to revalidate our BUFFER_CLIENT after releasing the buffer lock
       * because we may have been removed from the buffer by bm_cleanup() & co
       * due to a timeout or whatever. */
      pc = bm_get_my_client_locked(pbuf_guard);
 
      /* return if TCP connection broken */
      if (status == SS_ABORT)
         return SS_ABORT;
 
      if (timeout_msec == BM_NO_WAIT)
         return BM_ASYNC_RETURN;
   }
 
   if (pc->read_wait) {
      //printf("bm_wait_for_more_events: buffer [%s] client [%s] clear read_wait!\n", pheader->name, pc->name);
      pc->read_wait = FALSE;
   }
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_wakeup_producers_locked()

static void bm_wakeup_producers_locked ( const BUFFER_HEADER *  pheader, const BUFFER_CLIENT *  pc  )

static

Definition at line 8795 of file midas.cxx.

                                                                                              {
   int i;
   int have_get_all_requests = 0;
 
   for (i = 0; i < pc->max_request_index; i++)
      if (pc->event_request[i].valid)
         have_get_all_requests |= (pc->event_request[i].sampling_type == GET_ALL);
 
   /* only GET_ALL requests actually free space in the event buffer */
   if (!have_get_all_requests)
      return;
 
   /*
      If read pointer has been changed, it may have freed up some space
      for waiting producers. So check if free space is now more than 50%
      of the buffer size and wake waiting producers.
    */
 
   int free_space = pc->read_pointer - pheader->write_pointer;
   if (free_space <= 0)
      free_space += pheader->size;
 
   if (free_space >= pheader->size * 0.5) {
      for (i = 0; i < pheader->max_client_index; i++) {
         const BUFFER_CLIENT *pc = pheader->client + i;
         if (pc->pid && pc->write_wait) {
            BOOL send_wakeup = (pc->write_wait < free_space);
            //printf("bm_wakeup_producers: buffer [%s] client [%s] write_wait %d, free_space %d, sending wakeup message %d\n", pheader->name, pc->name, pc->write_wait, free_space, send_wakeup);
            if (send_wakeup) {
               ss_resume(pc->port, "B  ");
            }
         }
      }
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_write_buffer_statistics_to_odb()

static void bm_write_buffer_statistics_to_odb ( HNDLE  hDB, BUFFER *  pbuf, BOOL  force  )

static

Definition at line 6594 of file midas.cxx.

{
   //printf("bm_buffer_write_statistics_to_odb: buffer [%s] client [%s], lock count %d -> %d, force %d\n", pbuf->buffer_name, pbuf->client_name, pbuf->last_count_lock, pbuf->count_lock, force);
 
   bm_lock_buffer_guard pbuf_guard(pbuf);
 
   if (!pbuf_guard.is_locked())
      return;
 
   if (!force) {
      if (pbuf->count_lock == pbuf->last_count_lock) {
         return;
      }
   }
 
   std::string buffer_name = pbuf->buffer_name;
   std::string client_name = pbuf->client_name;
 
   if ((strlen(buffer_name.c_str()) < 1) || (strlen(client_name.c_str()) < 1)) {
      // do not call cm_msg() while holding buffer lock, if we are SYSMSG, we will deadlock. K.O.
      pbuf_guard.unlock(); // unlock before cm_msg()
      cm_msg(MERROR, "bm_write_buffer_statistics_to_odb", "Invalid empty buffer name \"%s\" or client name \"%s\"", buffer_name.c_str(), client_name.c_str());
      return;
   }
 
   pbuf->last_count_lock = pbuf->count_lock;
 
   BUFFER_INFO xbuf(pbuf);
   BUFFER_HEADER xheader = *pbuf->buffer_header;
   int client_index = pbuf->client_index;
 
   pbuf_guard.unlock();
 
   bm_write_buffer_statistics_to_odb_copy(hDB, buffer_name.c_str(), client_name.c_str(), client_index, &xbuf, &xheader);
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_write_buffer_statistics_to_odb_copy()

static void bm_write_buffer_statistics_to_odb_copy ( HNDLE  hDB, const char *  buffer_name, const char *  client_name, int  client_index, BUFFER_INFO *  pbuf, BUFFER_HEADER *  pheader  )

static

Definition at line 6472 of file midas.cxx.

{
   int status;
 
   DWORD now = ss_millitime();
 
   HNDLE hKey;
   status = db_find_key(hDB, 0, "/System/Buffers", &hKey);
   if (status != DB_SUCCESS) {
      db_create_key(hDB, 0, "/System/Buffers", TID_KEY);
      status = db_find_key(hDB, 0, "/System/Buffers", &hKey);
      if (status != DB_SUCCESS)
         return;
   }
 
   HNDLE hKeyBuffer;
   status = db_find_key(hDB, hKey, buffer_name, &hKeyBuffer);
   if (status != DB_SUCCESS) {
      db_create_key(hDB, hKey, buffer_name, TID_KEY);
      status = db_find_key(hDB, hKey, buffer_name, &hKeyBuffer);
      if (status != DB_SUCCESS)
         return;
   }
 
   double buf_size = pheader->size;
   double buf_rptr = pheader->read_pointer;
   double buf_wptr = pheader->write_pointer;
 
   double buf_fill = 0;
   double buf_cptr = 0;
   double buf_cused = 0;
   double buf_cused_pct = 0;
 
   if (client_index >= 0 && client_index <= pheader->max_client_index) {
      buf_cptr = pheader->client[client_index].read_pointer;
 
      if (buf_wptr == buf_cptr) {
         buf_cused = 0;
      } else if (buf_wptr > buf_cptr) {
         buf_cused = buf_wptr - buf_cptr;
      } else {
         buf_cused = (buf_size - buf_cptr) + buf_wptr;
      }
      
      buf_cused_pct = buf_cused / buf_size * 100.0;
      
      // we cannot write buf_cused and buf_cused_pct into the buffer statistics
      // because some other GET_ALL client may have different buf_cused & etc,
      // so they must be written into the per-client statistics
      // and the web page should look at all the GET_ALL clients and used
      // the biggest buf_cused as the whole-buffer "bytes used" value.
   }
 
   if (buf_wptr == buf_rptr) {
      buf_fill = 0;
   } else if (buf_wptr > buf_rptr) {
      buf_fill = buf_wptr - buf_rptr;
   } else {
      buf_fill = (buf_size - buf_rptr) + buf_wptr;
   }
 
   double buf_fill_pct = buf_fill / buf_size * 100.0;
 
   db_set_value(hDB, hKeyBuffer, "Size", &buf_size, sizeof(double), 1, TID_DOUBLE);
   db_set_value(hDB, hKeyBuffer, "Write pointer", &buf_wptr, sizeof(double), 1, TID_DOUBLE);
   db_set_value(hDB, hKeyBuffer, "Read pointer", &buf_rptr, sizeof(double), 1, TID_DOUBLE);
   db_set_value(hDB, hKeyBuffer, "Filled", &buf_fill, sizeof(double), 1, TID_DOUBLE);
   db_set_value(hDB, hKeyBuffer, "Filled pct", &buf_fill_pct, sizeof(double), 1, TID_DOUBLE);
 
   status = db_find_key(hDB, hKeyBuffer, "Clients", &hKey);
   if (status != DB_SUCCESS) {
      db_create_key(hDB, hKeyBuffer, "Clients", TID_KEY);
      status = db_find_key(hDB, hKeyBuffer, "Clients", &hKey);
      if (status != DB_SUCCESS)
         return;
   }
 
   HNDLE hKeyClient;
   status = db_find_key(hDB, hKey, client_name, &hKeyClient);
   if (status != DB_SUCCESS) {
      db_create_key(hDB, hKey, client_name, TID_KEY);
      status = db_find_key(hDB, hKey, client_name, &hKeyClient);
      if (status != DB_SUCCESS)
         return;
   }
 
   db_set_value(hDB, hKeyClient, "count_lock", &pbuf->count_lock, sizeof(int), 1, TID_INT32);
   db_set_value(hDB, hKeyClient, "count_sent", &pbuf->count_sent, sizeof(int), 1, TID_INT32);
   db_set_value(hDB, hKeyClient, "bytes_sent", &pbuf->bytes_sent, sizeof(double), 1, TID_DOUBLE);
   db_set_value(hDB, hKeyClient, "count_write_wait", &pbuf->count_write_wait, sizeof(int), 1, TID_INT32);
   db_set_value(hDB, hKeyClient, "time_write_wait", &pbuf->time_write_wait, sizeof(DWORD), 1, TID_UINT32);
   db_set_value(hDB, hKeyClient, "max_bytes_write_wait", &pbuf->max_requested_space, sizeof(INT), 1, TID_INT32);
   db_set_value(hDB, hKeyClient, "count_read", &pbuf->count_read, sizeof(int), 1, TID_INT32);
   db_set_value(hDB, hKeyClient, "bytes_read", &pbuf->bytes_read, sizeof(double), 1, TID_DOUBLE);
   db_set_value(hDB, hKeyClient, "get_all_flag", &pbuf->get_all_flag, sizeof(BOOL), 1, TID_BOOL);
   db_set_value(hDB, hKeyClient, "read_pointer", &buf_cptr, sizeof(double), 1, TID_DOUBLE);
   db_set_value(hDB, hKeyClient, "bytes_used", &buf_cused, sizeof(double), 1, TID_DOUBLE);
   db_set_value(hDB, hKeyClient, "pct_used", &buf_cused_pct, sizeof(double), 1, TID_DOUBLE);
 
   for (int i = 0; i < MAX_CLIENTS; i++) {
      if (!pbuf->client_count_write_wait[i])
         continue;
      
      if (pheader->client[i].pid == 0)
         continue;
      
      if (pheader->client[i].name[0] == 0)
         continue;
      
      char str[100 + NAME_LENGTH];
      
      sprintf(str, "writes_blocked_by/%s/count_write_wait", pheader->client[i].name);
      db_set_value(hDB, hKeyClient, str, &pbuf->client_count_write_wait[i], sizeof(int), 1, TID_INT32);
      
      sprintf(str, "writes_blocked_by/%s/time_write_wait", pheader->client[i].name);
      db_set_value(hDB, hKeyClient, str, &pbuf->client_time_write_wait[i], sizeof(DWORD), 1, TID_UINT32);
   }
 
   db_set_value(hDB, hKeyBuffer, "Last updated", &now, sizeof(DWORD), 1, TID_UINT32);
   db_set_value(hDB, hKeyClient, "last_updated", &now, sizeof(DWORD), 1, TID_UINT32);
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_write_statistics_to_odb()

INT bm_write_statistics_to_odb

(

void

)

Close all open buffers

Returns

BM_SUCCESS

Definition at line 7288 of file midas.cxx.

                                     {
#ifdef LOCAL_ROUTINES
   {
      int status;
      HNDLE hDB;
 
      status = cm_get_experiment_database(&hDB, NULL);
 
      if (status != CM_SUCCESS) {
         //printf("bm_write_statistics_to_odb: cannot get ODB handle!\n");
         return BM_SUCCESS;
      }
 
      std::vector<BUFFER*> mybuffers;
      
      gBuffersMutex.lock();
      mybuffers = gBuffers;
      gBuffersMutex.unlock();
 
      for (BUFFER* pbuf : mybuffers) {
         if (!pbuf || !pbuf->attached)
            continue;
         bm_write_buffer_statistics_to_odb(hDB, pbuf, FALSE);
      }
   }
#endif /* LOCAL_ROUTINES */
 
   return BM_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bm_write_to_buffer_locked()

static void bm_write_to_buffer_locked ( BUFFER_HEADER *  pheader, int  sg_n, const char *const  sg_ptr[], const size_t  sg_len[], size_t  total_size  )

static

Definition at line 9512 of file midas.cxx.

{
   char *pdata = (char *) (pheader + 1);
 
   //int old_write_pointer = pheader->write_pointer;
 
   /* new event fits into the remaining space? */
   if ((size_t)pheader->write_pointer + total_size <= (size_t)pheader->size) {
      //memcpy(pdata + pheader->write_pointer, pevent, event_size);
      char* wptr = pdata + pheader->write_pointer;
      for (int i=0; i<sg_n; i++) {
         //printf("memcpy %p+%d\n", sg_ptr[i], (int)sg_len[i]);
         memcpy(wptr, sg_ptr[i], sg_len[i]);
         wptr += sg_len[i];
      }
      pheader->write_pointer = pheader->write_pointer + total_size;
      assert(pheader->write_pointer <= pheader->size);
      /* remaining space is smaller than size of an event header? */
      if ((pheader->write_pointer + (int) sizeof(EVENT_HEADER)) > pheader->size) {
         // note: ">" here to match "bm_incr_rp". If remaining space is exactly
         // equal to the event header size, we will write the next event header here,
         // then wrap the pointer and write the event data at the beginning of the buffer.
         //printf("bm_write_to_buffer_locked: truncate wp %d. buffer size %d, remaining %d, event header size %d, event size %d, total size %d\n", pheader->write_pointer, pheader->size, pheader->size-pheader->write_pointer, (int)sizeof(EVENT_HEADER), event_size, total_size);
         pheader->write_pointer = 0;
      }
   } else {
      /* split event */
      size_t size = pheader->size - pheader->write_pointer;
 
      //printf("split: wp %d, size %d, avail %d\n", pheader->write_pointer, pheader->size, size);
 
      //memcpy(pdata + pheader->write_pointer, pevent, size);
      //memcpy(pdata, ((const char *) pevent) + size, event_size - size);
 
      char* wptr = pdata + pheader->write_pointer;
      size_t count = 0;
 
      // copy first part
 
      int i = 0;
      for (; i<sg_n; i++) {
         if (count + sg_len[i] > size)
            break;
         memcpy(wptr, sg_ptr[i], sg_len[i]);
         wptr  += sg_len[i];
         count += sg_len[i];
      }
 
      //printf("wptr %d, count %d\n", wptr-pdata, count);
 
      // split segment
 
      size_t first = size - count;
      size_t second = sg_len[i] - first;
      assert(first + second == sg_len[i]);
      assert(count + first == size);
      
      //printf("first %d, second %d\n", first, second);
      
      memcpy(wptr, sg_ptr[i], first);
      wptr = pdata + 0;
      count += first;
      memcpy(wptr, sg_ptr[i] + first, second);
      wptr  += second;
      count += second;
      i++;
 
      // copy remaining
 
      for (; i<sg_n; i++) {
         memcpy(wptr, sg_ptr[i], sg_len[i]);
         wptr  += sg_len[i];
         count += sg_len[i];
      }
 
      //printf("wptr %d, count %d\n", wptr-pdata, count);
 
      //printf("bm_write_to_buffer_locked: wrap wp %d -> %d. buffer size %d, available %d, wrote %d, remaining %d, event size %d, total size %d\n", pheader->write_pointer, total_size-size, pheader->size, pheader->size-pheader->write_pointer, size, pheader->size - (pheader->write_pointer+size), event_size, total_size);
 
      pheader->write_pointer = total_size - size;
   }
 
   //printf("bm_write_to_buffer_locked: buf [%s] size %d, wrote %d/%d, wp %d -> %d\n", pheader->name, pheader->size, event_size, total_size, old_write_pointer, pheader->write_pointer);
}

Here is the call graph for this function:

Here is the caller graph for this function:

Variable Documentation

_bm_lock_timeout

int _bm_lock_timeout = 5 * 60 * 1000

static

Definition at line 5927 of file midas.cxx.

_bm_max_event_size

DWORD _bm_max_event_size = 0

static

Definition at line 5922 of file midas.cxx.

_bm_mutex_timeout_sec

double _bm_mutex_timeout_sec = _bm_lock_timeout/1000 + 15.000

static

Definition at line 5928 of file midas.cxx.

defrag_buffer

EVENT_DEFRAG_BUFFER defrag_buffer[MAX_DEFRAG_EVENTS]

static

Definition at line 11289 of file midas.cxx.