octopi/library/octopus/entity_data_bin.cpp

   1 /*****************************************************************************\
   2 *                                                                             *
   3 *  Name   : entity_data_bin                                                   *
   4 *  Author : Chris Koeritz                                                     *
   5 *                                                                             *
   6 *******************************************************************************
   7 * Copyright (c) 2002-$now By Author.  This program is free software; you can  *
   8 * redistribute it and/or modify it under the terms of the GNU General Public  *
   9 * License as published by the Free Software Foundation; either version 2 of   *
  10 * the License or (at your option) any later version.  This is online at:      *
  11 *     http://www.fsf.org/copyleft/gpl.html                                    *
  12 * Please send any updates to: fred@gruntose.com                               *
  13 \*****************************************************************************/
  14
  15 #include "entity_data_bin.h"
  16 #include "entity_defs.h"
  17 #include "infoton.h"
  18 #include "tentacle.h"
  19
  20 #include <basis/astring.h>
  21
  22 #include <basis/mutex.h>
  23 #include <loggers/program_wide_logger.h>
  24 #include <structures/set.h>
  25 #include <structures/string_array.h>
  26 #include <structures/amorph.h>
  27 #include <structures/string_hash.h>
  28 #include <textual/parser_bits.h>
  29 #include <timely/time_stamp.h>
  30
  31 using namespace basis;
  32 using namespace loggers;
  33 using namespace structures;
  34 using namespace textual;
  35 using namespace timely;
  36
  37 namespace octopi {
  38
  39 //#define DEBUG_ENTITY_DATA_BIN
  40   // uncomment for more debugging information.
  41
  42 #undef GRAB_LOCK
  43 #define GRAB_LOCK \
  44   auto_synchronizer l(*_ent_lock)
  45
  46 #undef LOG
  47 #define LOG(s) CLASS_EMERGENCY_LOG(program_wide_logger::get(), s)
  48
  49 const int OCTOPUS_TABLE_BITS = 6;
  50   // the hash table for items will have 2^N entries.
  51
  52 //hmmm: parameterize in class interface?
  53 ////const int DATA_DECAY_INTERVAL = 4 * MINUTE_ms;
  54   // if we haven't gotten a data item out to its entity in this long, then
  55   // we assume the entity has croaked or doesn't want its data.
  56
  57 //////////////
  58
  59 class infoton_holder
  60 {
  61 public:
  62   infoton *_item;      // the data making up the production.
  63   octopus_request_id _id;  // the id, if any, of the original request.
  64   time_stamp _when_added;  // when the data became available.
  65
  66   infoton_holder(const octopus_request_id &id = octopus_request_id(),
  67       infoton *item = NIL)
  68   : _item(item), _id(id), _when_added() {}
  69
  70   ~infoton_holder() { WHACK(_item); }
  71
  72   astring text_form() const {
  73     return astring("id=") + _id.text_form() + ", added="
  74         + _when_added.text_form() + ", item="
  75         + _item->classifier().text_form() + ", data="
  76         + _item->text_form();
  77   }
  78 };
  79
  80 //////////////
  81
  82 class entity_basket : public amorph<infoton_holder>
  83 {
  84 public:
  85   time_stamp _last_active;
  86
  87   astring text_form() const {
  88     astring to_return;
  89     for (int i = 0; i < elements(); i++)
  90       to_return += get(i)->text_form() + parser_bits::platform_eol_to_chars();
  91     return to_return;
  92   }
  93 };
  94
  95 //////////////
  96
  97 class entity_hasher : public hashing_algorithm
  98 {
  99 public:
 100   virtual hashing_algorithm *clone() const { return new entity_hasher; }
 101
 102   virtual basis::un_int hash(const void *key_data, int formal(key_length)) const {
 103     octopus_entity *key = (octopus_entity *)key_data;
 104     // jiggle the pieces of the id into a number.
 105     return basis::un_int(
 106         key->process_id()
 107         + (key->add_in() << 10)
 108         + (key->sequencer() << 14)
 109         + (key->hostname()[0] << 20)
 110         + (key->hostname()[1] << 24) );
 111   }
 112 };
 113
 114 //////////////
 115
 116 class entity_item_hash
 117 : public hash_table<octopus_entity, entity_basket>
 118 {
 119 public:
 120   entity_item_hash(const entity_hasher &hash)
 121   : hash_table<octopus_entity, entity_basket>(hash, OCTOPUS_TABLE_BITS)
 122   {}
 123 };
 124
 125 //////////////
 126
 127 class basketcase : public structures::set<octopus_entity>
 128 {
 129 public:
 130 };
 131
 132 //////////////
 133
 134 // used for our apply methods for communicating back to the caller.
 135 struct apply_struct
 136 {
 137   basketcase *_empty_baskets;
 138   entity_basket *_any_item;
 139   int &_items_held;  // hooks to parent's item count.
 140   int _decay_interval;  // how long are items allowed to live?
 141
 142   apply_struct(int &items_held)
 143       : _empty_baskets(NIL), _any_item(NIL), _items_held(items_held),
 144         _decay_interval(0) {}
 145 };
 146
 147 //////////////
 148
 149 entity_data_bin::entity_data_bin(int max_size_per_entity)
 150 : _table(new entity_item_hash(entity_hasher())),
 151   _ent_lock(new mutex),
 152   _action_count(0),
 153   _max_per_ent(max_size_per_entity),
 154   _items_held(0)
 155 {}
 156
 157 entity_data_bin::~entity_data_bin()
 158 {
 159   WHACK(_table);
 160   WHACK(_ent_lock);
 161 }
 162
 163 int entity_data_bin::entities() const
 164 {
 165   GRAB_LOCK;
 166   return _table->elements();
 167 }
 168
 169 struct text_form_accumulator { astring _accum; };
 170
 171 bool text_form_applier(const octopus_entity &formal(key), entity_basket &bask,
 172     void *data_link)
 173 {
 174   text_form_accumulator *shuttle = (text_form_accumulator *)data_link;
 175   shuttle->_accum += bask.text_form();
 176   return true;
 177 }
 178
 179 astring entity_data_bin::text_form() const
 180 {
 181   GRAB_LOCK;
 182   text_form_accumulator shuttle;
 183   _table->apply(text_form_applier, &shuttle);
 184   return shuttle._accum;
 185 }
 186
 187 bool scramble_applier(const octopus_entity &formal(key), entity_basket &bask,
 188     void *data_link)
 189 {
 190   #undef static_class_name
 191   #define static_class_name() "entity_data_bin"
 192   FUNCDEF("scramble_applier");
 193   int *county = (int *)data_link;
 194   *county += bask.elements();
 195   return true;
 196   #undef static_class_name
 197 }
 198
 199 // this could be extended to do more interesting checks also; currently it's
 200 // just like the entities() method really.
 201 int entity_data_bin::scramble_counter()
 202 {
 203   GRAB_LOCK;
 204   int count = 0;
 205   _table->apply(scramble_applier, &count);
 206   return count;
 207 }
 208
 209 #ifdef DEBUG_ENTITY_DATA_BIN
 210   #define DUMP_STATE \
 211     if ( !(_action_count++ % 100) ) { \
 212       int items = scramble_counter(); \
 213       LOG(a_sprintf("-> %d items counted.", items)); \
 214     }
 215 #else
 216   #define DUMP_STATE
 217 #endif
 218
 219 bool entity_data_bin::add_item(infoton *to_add,
 220     const octopus_request_id &orig_id)
 221 {
 222   FUNCDEF("add_item");
 223   GRAB_LOCK;
 224   // create a record to add to the appropriate bin.
 225   infoton_holder *holder = new infoton_holder(orig_id, to_add);
 226
 227   // see if a basket already exists for the entity.
 228   entity_basket *bask = _table->find(orig_id._entity);
 229   if (!bask) {
 230     // this entity doesn't have a basket so add one.
 231     bask = new entity_basket;
 232     _table->add(orig_id._entity, bask);
 233   }
 234
 235   bask->_last_active = time_stamp();  // reset activity time.
 236
 237   // count up the current amount of data in use.
 238   int current_size = 0;
 239   for (int i = 0; i < bask->elements(); i++)
 240     current_size += bask->borrow(i)->_item->packed_size();
 241
 242   if (current_size + to_add->packed_size() > _max_per_ent) {
 243     WHACK(holder);
 244     return false;
 245   }
 246
 247   // append the latest production to the list.
 248   bask->append(holder);
 249   _items_held++;
 250   return true;
 251 }
 252
 253 bool any_item_applier(const octopus_entity &formal(key), entity_basket &bask,
 254     void *data_link)
 255 {
 256   #define static_class_name() "entity_data_bin"
 257   FUNCDEF("any_item_applier");
 258   apply_struct *apple = (apply_struct *)data_link;
 259   // check the basket to see if it has any items.
 260   if (!bask.elements()) {
 261 //#ifdef DEBUG_ENTITY_DATA_BIN
 262 //    LOG(astring("saw empty basket ") + key.mangled_form());
 263 //#endif
 264     return true;  // continue iterating.
 265   }
 266   apple->_any_item = &bask;
 267   return false;  // stop iteration.
 268   #undef static_class_name
 269 }
 270
 271 infoton *entity_data_bin::acquire_for_any(octopus_request_id &id)
 272 {
 273   FUNCDEF("acquire_for_any");
 274   GRAB_LOCK;
 275   apply_struct apple(_items_held);
 276   _table->apply(any_item_applier, &apple);
 277   if (!apple._any_item) return NIL;
 278   DUMP_STATE;
 279   // retrieve the information from our basket that was provided.
 280   infoton_holder *found = apple._any_item->acquire(0);
 281   apple._any_item->zap(0, 0);
 282   if (!apple._any_item->elements()) {
 283     // toss this empty basket.
 284 #ifdef DEBUG_ENTITY_DATA_BIN
 285     LOG(astring("tossing empty basket ") + found->_id._entity.mangled_form());
 286 #endif
 287     _table->zap(found->_id._entity);
 288   }
 289   apple._any_item = NIL;
 290   infoton *to_return = found->_item;
 291   id = found->_id;
 292   found->_item = NIL;  // clear so it won't be whacked.
 293   WHACK(found);
 294   _items_held--;
 295 //#ifdef DEBUG_ENTITY_DATA_BIN
 296   if (_items_held < 0)
 297     LOG("logic error: number of items went below zero.");
 298 //#endif
 299   return to_return;
 300 }
 301
 302 int entity_data_bin::acquire_for_entity(const octopus_entity &requester,
 303     infoton_list &items, int maximum_size)
 304 {
 305   FUNCDEF("acquire_for_entity [multiple]");
 306   // this method does not grab the lock because it simply composes other
 307   // class methods without interacting with class data members.
 308   items.reset();
 309   if (maximum_size <= 0) maximum_size = 20 * KILOBYTE;
 310     // pick a reasonable default.
 311   octopus_request_id id;
 312   int items_found = 0;
 313   while (maximum_size > 0) {
 314     infoton *inf = acquire_for_entity(requester, id);
 315     if (!inf)
 316       break;  // none left.
 317     items.append(new infoton_id_pair(inf, id));
 318     maximum_size -= inf->packed_size();
 319     items_found++;
 320   }
 321   return items_found;
 322 }
 323
 324 infoton *entity_data_bin::acquire_for_entity(const octopus_entity &requester,
 325     octopus_request_id &id)
 326 {
 327   FUNCDEF("acquire_for_entity [single]");
 328   id = octopus_request_id();  // reset it.
 329   GRAB_LOCK;
 330   infoton *to_return = NIL;
 331   entity_basket *bask = _table->find(requester);
 332   if (!bask) {
 333     return NIL;
 334   }
 335   if (!bask->elements()) {
 336 #ifdef DEBUG_ENTITY_DATA_BIN
 337     LOG(astring("tossing empty basket ") + requester.mangled_form());
 338 #endif
 339     _table->zap(requester);
 340     return NIL;
 341   }
 342   DUMP_STATE;
 343   id = bask->get(0)->_id;
 344   to_return = bask->borrow(0)->_item;
 345   bask->borrow(0)->_item = NIL;
 346   bask->zap(0, 0);
 347   if (!bask->elements()) {
 348 #ifdef DEBUG_ENTITY_DATA_BIN
 349     LOG(astring("tossing empty basket ") + requester.mangled_form());
 350 #endif
 351     _table->zap(requester);
 352   }
 353   _items_held--;
 354 //#ifdef DEBUG_ENTITY_DATA_BIN
 355   if (_items_held < 0)
 356     LOG("logic error: number of items went below zero.");
 357 //#endif
 358   return to_return;
 359 }
 360
 361 infoton *entity_data_bin::acquire_for_identifier(const octopus_request_id &id)
 362 {
 363   FUNCDEF("acquire_for_identifier");
 364   infoton *to_return = NIL;
 365   GRAB_LOCK;
 366   entity_basket *bask = _table->find(id._entity);
 367   if (!bask) return NIL;
 368   if (!bask->elements()) {
 369 #ifdef DEBUG_ENTITY_DATA_BIN
 370     LOG(astring("tossing empty basket ") + id._entity.mangled_form());
 371 #endif
 372     _table->zap(id._entity);
 373     return NIL;
 374   }
 375   for (int i = 0; i < bask->elements(); i++) {
 376     if (bask->get(i)->_id == id) {
 377       to_return = bask->borrow(i)->_item;  // snag the item.
 378       bask->borrow(i)->_item = NIL;  // clear the list's version out.
 379       bask->zap(i, i);  // whack the sanitized element.
 380       DUMP_STATE;
 381       if (!bask->elements()) {
 382 #ifdef DEBUG_ENTITY_DATA_BIN
 383         LOG(astring("tossing empty basket ") + id._entity.mangled_form());
 384 #endif
 385         _table->zap(id._entity);
 386       }
 387       _items_held--;
 388 //#ifdef DEBUG_ENTITY_DATA_BIN
 389       if (_items_held < 0)
 390         LOG("logic error: number of items went below zero.");
 391 //#endif
 392       return to_return;
 393     }
 394   }
 395   return NIL;
 396 }
 397
 398 bool cleaning_applier(const octopus_entity &key, entity_basket &bask,
 399     void *data_link)
 400 {
 401   #define static_class_name() "entity_data_bin"
 402   FUNCDEF("cleaning_applier");
 403   apply_struct *apple = (apply_struct *)data_link;
 404   time_stamp expiration_time(-apple->_decay_interval);
 405
 406   int whack_count = 0;
 407   for (int i = 0; i < bask.elements(); i++) {
 408     infoton_holder &rec = *bask.borrow(i);
 409     if (rec._when_added <= expiration_time) {
 410       // if a requester hasn't picked this up in N seconds, then drop it.
 411 #ifdef DEBUG_ENTITY_DATA_BIN
 412       LOG(astring("whacking old item ") + rec._id.text_form());
 413 #endif
 414       whack_count++;
 415       apple->_items_held--;
 416 //#ifdef DEBUG_ENTITY_DATA_BIN
 417       if (apple->_items_held < 0)
 418         LOG("logic error: number of items went below zero.");
 419 //#endif
 420       bask.zap(i, i);
 421       i--;  // skip back before the delete.
 422     } else {
 423       // NOTE: this break is based on an assumption about the storage of
 424       // items; if it's ever the case in the future that items can be
 425       // disordered on time of arrival in the queue, then the break should
 426       // be removed.
 427       break;
 428     }
 429   }
 430 #ifdef DEBUG_ENTITY_DATA_BIN
 431   if (whack_count)
 432     LOG(a_sprintf("==> whacked %d old items.", whack_count));
 433 #endif
 434   if (!bask.elements()) {
 435     // if the basket has nothing left in it then we signal the parent that
 436     // it can be deleted.
 437 //LOG("adding to empty basket list.");
 438     *apple->_empty_baskets += key;
 439 //LOG("added to empty basket list.");
 440   }
 441
 442   // keep iterating on items unless we know it's time to go.
 443   return true;
 444   #undef static_class_name
 445 }
 446
 447 void entity_data_bin::clean_out_deadwood(int decay_interval)
 448 {
 449 #ifdef DEBUG_ENTITY_DATA_BIN
 450   FUNCDEF("clean_out_deadwood");
 451 #endif
 452   GRAB_LOCK;
 453   // check that no items have timed out.
 454   apply_struct apple(_items_held);
 455   basketcase empty_baskets;
 456   apple._empty_baskets = &empty_baskets;
 457   apple._decay_interval = decay_interval;
 458   _table->apply(cleaning_applier, &apple);
 459
 460   // clean up any entities whose baskets are empty.
 461   for (int i = empty_baskets.length() - 1; i >= 0; i--) {
 462 #ifdef DEBUG_ENTITY_DATA_BIN
 463      LOG(astring("removing basket ") + empty_baskets.get(i).mangled_form());
 464 #endif
 465     _table->zap(empty_baskets.get(i));
 466     empty_baskets.zap(i, i);
 467     // we don't skip back since we're scanning the array from its end.
 468   }
 469 }
 470
 471 bool entity_data_bin::get_sizes(const octopus_entity &id, int &items,
 472     int &bytes)
 473 {
 474   FUNCDEF("get_sizes");
 475   items = 0;
 476   bytes = 0;
 477   GRAB_LOCK;
 478   entity_basket *bask = _table->find(id);
 479   if (!bask || !bask->elements()) return false;
 480   items = bask->elements();
 481   for (int i = 0; i < bask->elements(); i++)
 482     bytes += bask->borrow(i)->_item->packed_size();
 483   return true;
 484 }
 485
 486 } //namespace.
 487