zing_table.cpp

Go to the documentation of this file.

#ifndef ZING_TABLE_IMPLEMENTATION_FILE
#define ZING_TABLE_IMPLEMENTATION_FILE

/*****************************************************************************\
*                                                                             *
*  Name   : zing_table                                                        *
*  Author : Chris Koeritz                                                     *
*                                                                             *
*******************************************************************************
* Copyright (c) 1996-$now By Author.  This program is free software; you can  *
* redistribute it and/or modify it under the terms of the GNU General Public  *
* License as published by the Free Software Foundation; either version 2 of   *
* the License or (at your option) any later version.  This is online at:      *
*     http://www.fsf.org/copyleft/gpl.html                                    *
* Please send any updates to: fred@gruntose.com                               *
\*****************************************************************************/

#include "event_record.h"
#include "zing_table.h"

#include <basis/auto_synch.h>
#include <basis/mutex.h>
#include <data_struct/amorph.cpp>
#include <data_struct/int_hash.cpp>
#include <data_struct/memory_limiter.h>
#include <mechanisms/heartbeat.h>
#include <mechanisms/time_stamp.h>
#include <opsystem/system_values.h>
#include <textual/string_manipulation.h>

using namespace basis;

//#define DEBUG_ZING_TABLE
  // uncomment if you want noisy zinging.

#undef LOG
#define LOG(s) CLASS_EMERGENCY_LOG(program_wide_logger(), s)

const int ZING_COMPLAINT_INTERVAL = 2 * MINUTE_ms;
  // the frequency at which we complain about overflows on zinging.


class queued_zings : public amorph<event_record> {};


class zing_queue_amorph : public int_hash<queued_zings>
{
public:
  zing_queue_amorph() : int_hash<queued_zings>(10) {}
    // 1024 has slots.
//hmmm: is that too many?
};


zing_table::zing_table(int overall_memory, int per_user_memory)
: _clients(new zing_queue_amorph),
  _lock(new mutex),
  _limiter(new memory_limiter(overall_memory, per_user_memory))
{}

zing_table::~zing_table()
{
  WHACK(_lock);
  WHACK(_clients);
  WHACK(_limiter);
}

bool zing_table::add_client(int new_id)
{
  auto_synchronizer l(*_lock);
  queued_zings *cli = _clients->find(new_id);
  if (cli) return false;  // already exists.
  queued_zings *new_entry = new queued_zings;
  _clients->add(new_id, new_entry);
//check result if it matters whether this was new or existing.
  return true;
}

bool zing_table::remove_client(int outgoing_id)
{
  auto_synchronizer l(*_lock);
  queued_zings *cli = _clients->find(outgoing_id);
  if (!cli) return false;  // not found for that id.
  _clients->zap(outgoing_id);
  return true;
}

void zing_table::zing_event(int client_id, int event, const byte_array &data)
{
  zing_event(client_id, event, data, int_set());
}

void zing_table::zing_event(int client_id, int event, const byte_array &data,
    const int_set &targets)
{
  FUNCDEF("zing_event");
  auto_synchronizer l(*_lock);

/*
{
istring targs;
for (int n = 0; n < targets.elements(); n++)
targs += isprintf("%d ", targets[n]);
if (targs.t()) LOG(istring("targets passed in: ") + targs);
}
*/

  // we always repack the data in the event structure because that's the
  // known protocol.  no one should be pre-packing these structures.
  event_record to_add(client_id, event, data, targets);
  byte_array packed_blob;
  to_add.pack(packed_blob);

  int size_guess = sizeof(event_record) + packed_blob.length()
      + 10 * sizeof(int) + (targets.length() + 1) * sizeof(int);
  if (!_limiter->okay_allocation(client_id, size_guess)) {
    // complain that we're not allowing this, but not too frequently.
static time_stamp last_complaint(-ZING_COMPLAINT_INTERVAL * 2);
//move to object header; safe for now since we're synched.
    if (time_stamp().value() - last_complaint.value()
        > ZING_COMPLAINT_INTERVAL) {
      LOG(isprintf("data overflow on client %d!", client_id));
      last_complaint = time_stamp();
    }
    return;
  }

  queued_zings *cab = _clients->find(client_id);
  if (!cab) return;  // not found for that id.

  event_record *to_store
      = new event_record(client_id, event, packed_blob, targets);
  cab->append(to_store);
}

int zing_table::show_events(int client_id, const istring &name,
    istring &output, int indentation)
{
  istring indent = string_manipulation::indentation(indentation);
  int events_printed = 0;

  auto_synchronizer l(*_lock);
  queued_zings *cab = _clients->find(client_id);
  if (!cab) return 0;  // not found for that id.

  system_values event_namer(system_values::EVENT_VALUES());
  for (int i = 0; i < cab->elements(); i++) {
    event_record *w = cab->borrow(i);
    if (w) {
      istring event_name;
      istring descrip;
      istring file_loc;
      event_namer.lookup(w->_event, event_name, descrip, file_loc);
      // finds a name for the event, if possible.
      if (!event_name)
        event_name = isprintf("%d", w->_event);
      output += indent + isprintf("%s with id %d has an event \"%s\" "
          "listed (%d bytes).", name.s(), client_id, event_name.s(),
          w->_data.length()) + log_base::platform_ending();
      events_printed++;
    }
  }
  return events_printed;
}

bool zing_table::pop_event(int client_id, int &event, byte_array &data,
    int_set &targets)
{
  FUNCDEF("pop_event");
  event = 0;
  data = NIL;

  auto_synchronizer l(*_lock);
  queued_zings *cab = _clients->find(client_id);
  if (!cab) return false;  // not found for that id.

  if (!cab->elements()) return false;
  event_record *w = cab->acquire(0);
  cab->zap(0, 0);
  if (w) {
    event = w->_event;
    data = w->_data;
    targets = w->_targets;

    int size_guess = sizeof(event_record) + w->_data.length()
        + 10 * sizeof(int) + (w->_targets.length() + 1) * sizeof(int);
    if (!_limiter->record_deletion(client_id, size_guess)) {
      LOG("logic error in zing size assessment.");
    }

    return true;
  }
  return false;
}

bool zing_table::peek_event(int client_id, int &event, int &data_length,
    int_set &targets) const
{
  event = 0;
  data_length = 0;

  auto_synchronizer l(*_lock);
  queued_zings *cab = _clients->find(client_id);
  if (!cab) return false;  // not found for that id.

  const event_record *w = cab->borrow(0);
  if (w) {
    event = w->_event;
    data_length = w->_data.length();
    targets = w->_targets;
    return true;
  }
  return false;
}

void zing_table::flush_events(int client_id)
{
  auto_synchronizer l(*_lock);
  queued_zings *cab = _clients->find(client_id);
  if (!cab) return;  // not found for that id.
  cab->reset();
}

bool zing_table::pop_events(int client_id, int &count, byte_array &event_data,
    int max_size)
{
  FUNCDEF("pop_events");
#ifdef DEBUG_ZINGABLE
  LOG(istring(istring::SPRINTF, "coming in with max len of %d.", max_size));
#endif
  count = 0;

//hmmm, what about grabbing all of it now?

  queued_zings data_holder;
  {
    auto_synchronizer l(*_lock);
    queued_zings *cab = _clients->find(client_id);
    if (!cab) return false;  // not found for that id.
    if (!cab->elements()) return false;  // nothing waiting.
    // move all pending items into the holder.
    for (int i = 0; i < cab->elements(); i++) {
      data_holder.append(cab->acquire(i));
    }
    cab->reset();  // dump all index storage.
  }

  int holder = 0;
  int start_position = event_data.length();  // where we're adding.
  bool added_holder = false;  // did we add the holder yet?
  int index;  // track the position so we can clear what we consumed.
  for (index = 0; index < data_holder.elements(); index++) {
    event_record *found = data_holder.borrow(index);
    if (!found) {
      data_holder.zap(index, index);
      index--;  // skip back for the missing one.
      continue;
    }
    // if we didn't attach the holder yet, we need to consider that we will.
    int extra_add = added_holder? 0 : sizeof(int);
    if (max_size
        && (extra_add + event_data.length() + found->_data.length()
            > max_size))
      break;
    if (!added_holder) {
      attach(event_data, holder);
      added_holder = true;
    }
    count++;
#ifdef DEBUG_ZINGABLE
    LOG(isprintf("for cli %d: adding event=%d cli=%d ev=%d len=%d.",
        client_id, count, found->_id, found->_event, found->_data.length()));
#endif
    event_data += found->_data;

//hmmm: move this into the ipc event structure object!
    int size_guess = sizeof(event_record) + found->_data.length()
        + 10 * sizeof(int) + (found->_targets.length() + 1) * sizeof(int);
    {
      auto_synchronizer l(*_lock);
      if (!_limiter->record_deletion(client_id, size_guess)) {
        LOG("logic error in zing size assessment.");
      }
    }
  }
  if (added_holder) {
    // store the number of events over the placeholder, if one's in there.
    byte_array to_replace_holder;
    attach(to_replace_holder, count);
    event_data.overwrite(start_position, to_replace_holder);
  }
  // if we chowed anything, remove it from the list.
  if (index)
    data_holder.zap(0, index - 1);

  // if we didn't clear the list, then we need to put the extras back.
  if (data_holder.elements()) {
    auto_synchronizer l(*_lock);
    queued_zings *cab = _clients->find(client_id);
    if (cab) {
      // move all remaining pending items back into the cabinet.
      cab->insert(0, data_holder.elements());
        // add space for these items.
      for (int i = 0; i < data_holder.elements(); i++) {
        cab->put(i, data_holder.acquire(i));
      }
    }
    data_holder.reset();  // clear for the heck of it.
  }

  return count > 0;
}

bool zing_table::peek_events(int client_id, int &count,
    int &data_length) const
{
  count = 0;
  data_length = 0;

  auto_synchronizer l(*_lock);
  queued_zings *cab = _clients->find(client_id);
  if (!cab) return false;  // not found for that id.

  if (!cab->elements()) return false;  // nothing waiting.
  data_length = sizeof(int);  // initially just the count size.
  for (int i = 0; i < cab->elements(); i++) {
    const event_record *found = cab->borrow(i);
    if (found) {
      count++;
      data_length += found->_data.length();
    }
  }
  return count > 0;
}


#endif //ZING_TABLE_IMPLEMENTATION_FILE

---