octopi/library/tests_octopus/t_bin_threaded.cpp

   1 /*****************************************************************************\
   2 *                                                                             *
   3 *  Name   : test_entity_data_bin_threaded                                     *
   4 *  Author : Chris Koeritz                                                     *
   5 *                                                                             *
   6 *******************************************************************************
   7 * Copyright (c) 2010-$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 <application/hoople_main.h>
  16 #include <basis/byte_array.h>
  17 #include <mathematics/chaos.h>
  18 #include <basis/functions.h>
  19 #include <basis/guards.h>
  20 #include <basis/astring.h>
  21 #include <basis/mutex.h>
  22 #include <structures/amorph.h>
  23 #include <structures/static_memory_gremlin.h>
  24 #include <loggers/console_logger.h>
  25 #include <processes/ethread.h>
  26 #include <processes/safe_roller.h>
  27 #include <timely/time_control.h>
  28 #include <timely/time_stamp.h>
  29 #include <octopus/entity_data_bin.h>
  30 #include <octopus/entity_defs.h>
  31 #include <octopus/unhandled_request.h>
  32 #include <application/application_shell.h>
  33 #include <configuration/application_configuration.h>
  34 #include <textual/string_manipulation.h>
  35
  36 #include <stdio.h>
  37
  38 #ifdef __WIN32__
  39   #include <process.h>
  40 #endif
  41
  42 using namespace application;
  43 using namespace loggers;
  44 using namespace octopi;
  45 using namespace processes;
  46 using namespace timely;
  47
  48 // global constants...
  49
  50 // how much data is the entity data bin allowed to hold at one time.
  51 const int MAXIMUM_DATA_PER_ENTITY = 1 * KILOBYTE;
  52 //tiny limit to test having too much data.
  53
  54 // controls the timing of the thread that adds items.
  55 const int MIN_ADDER_THREAD_PAUSE = 3;
  56 const int MAX_ADDER_THREAD_PAUSE = 20;
  57
  58 // controls the timing of the item deleting thread.
  59 const int MIN_WHACKER_THREAD_PAUSE = 8;
  60 const int MAX_WHACKER_THREAD_PAUSE = 70;
  61
  62 // bound the randomly chosen pause time for the cleanup thread.
  63 const int MIN_TIDIER_THREAD_PAUSE = 60;
  64 const int MAX_TIDIER_THREAD_PAUSE = 500;
  65
  66 // monk is kept asleep most of the time or he'd be trashing
  67 // all our data too frequently.
  68 const int MIN_MONK_THREAD_PAUSE = 2 * MINUTE_ms;
  69 const int MAX_MONK_THREAD_PAUSE = 4 * MINUTE_ms;
  70
  71 // the range of new items added whenever the creator thread is hit.
  72 const int MINIMUM_ITEMS_ADDED = 1;
  73 const int MAXIMUM_ITEMS_ADDED = 20;
  74
  75 const int DEFAULT_THREADS = 90;
  76   // the number of threads we create by default.
  77
  78 const int DEFAULT_RUN_TIME = 80 * MINUTE_ms;
  79 //2 * MINUTE_ms;
  80   // the length of time to run the program.
  81
  82 const int DATA_DECAY_TIME = 1 * MINUTE_ms;
  83   // how long we retain unclaimed data.
  84
  85 const int MONKS_CLEANING_TIME = 10 * SECOND_ms;
  86   // a very short duration for data to live.
  87
  88 #define LOG(to_print) printf("%s\n", (char *)astring(to_print).s());
  89 //CLASS_EMERGENCY_LOG(program_wide_logger().get(), to_print)
  90   // our macro for logging with a timestamp.
  91
  92 // global objects...
  93
  94 chaos _rando;  // our randomizer.
  95
  96 // replace app_shell version with local randomizer, so all the static
  97 // functions can employ it also.
  98 #define randomizer() _rando
  99
 100 entity_data_bin binger(MAXIMUM_DATA_PER_ENTITY);
 101
 102 octopus_request_id create_request_id()
 103 {
 104   // test the basic filling of the values in an entity.
 105   octopus_request_id req_id;
 106   if (randomizer().inclusive(1, 100) < 25) {
 107     // some of the time we make a totally random entity id.
 108     int sequencer = randomizer().inclusive(1, MAXINT - 10);
 109     int add_in = randomizer().inclusive(0, MAXINT - 10);
 110     int process_id = randomizer().inclusive(0, MAXINT - 10);
 111     req_id._entity = octopus_entity(string_manipulation::make_random_name(),
 112         process_id, sequencer, add_in);
 113   } else {
 114     // sometimes we use a less random identity.
 115     int sequencer = randomizer().inclusive(1, 3);
 116     int add_in = 12;
 117     int process_id = randomizer().inclusive(1, 4);
 118     req_id._entity = octopus_entity("boringentity",
 119         process_id, sequencer, add_in);
 120   }
 121   req_id._request_num = randomizer().inclusive(1, MAXINT - 10);
 122   return req_id;
 123 }
 124
 125 // this thread creates new items for the entity data bin.
 126 class ballot_box_stuffer : public ethread
 127 {
 128 public:
 129   ballot_box_stuffer() : ethread(0) {
 130     FUNCDEF("constructor");
 131     LOG("+creator");
 132   }
 133
 134   virtual ~ballot_box_stuffer() {
 135     FUNCDEF("destructor");
 136     LOG("~creator");
 137   }
 138
 139   DEFINE_CLASS_NAME("ballot_box_stuffer");
 140
 141   void perform_activity(void *formal(data)) {
 142     FUNCDEF("perform_activity");
 143     while (!should_stop()) {
 144       // add a new item to the cache.
 145       int how_many = randomizer().inclusive(MINIMUM_ITEMS_ADDED,
 146           MAXIMUM_ITEMS_ADDED);
 147       for (int i = 0; i < how_many; i++) {
 148         string_array random_strings;
 149         int string_count = randomizer().inclusive(1, 10);
 150         // we create a random classifier, just to use up some space.
 151         for (int q = 0; q < string_count; q++) {
 152           random_strings += string_manipulation::make_random_name();
 153         }
 154         unhandled_request *newbert = new unhandled_request(create_request_id(),
 155             random_strings);
 156         binger.add_item(newbert, create_request_id());
 157       }
 158       // snooze.
 159       int sleepy_time = randomizer().inclusive(MIN_ADDER_THREAD_PAUSE,
 160           MAX_ADDER_THREAD_PAUSE);
 161       time_control::sleep_ms(sleepy_time);
 162     }
 163   }
 164
 165 };
 166
 167 // this thread eliminates entries in the ballot box.
 168 class vote_destroyer : public ethread
 169 {
 170 public:
 171   vote_destroyer() : ethread(0) {
 172     FUNCDEF("constructor");
 173     LOG("+destroyer");
 174   }
 175
 176   virtual ~vote_destroyer() {
 177     FUNCDEF("destructor");
 178     LOG("~destroyer");
 179   }
 180
 181   DEFINE_CLASS_NAME("vote_destroyer");
 182
 183   void perform_activity(void *formal(data)) {
 184     FUNCDEF("perform_activity");
 185     while (!should_stop()) {
 186       // snag any old item and drop it on the floor.
 187       octopus_request_id id;
 188       infoton *found = binger.acquire_for_any(id);
 189       WHACK(found);
 190       // snooze.
 191       int sleepy_time = randomizer().inclusive(MIN_WHACKER_THREAD_PAUSE,
 192           MAX_WHACKER_THREAD_PAUSE);
 193       time_control::sleep_ms(sleepy_time);
 194     }
 195   }
 196 };
 197
 198 // this class makes sure the deadwood is cleaned out of the entity bin.
 199 class obsessive_compulsive : public ethread
 200 {
 201 public:
 202   obsessive_compulsive() : ethread(0) {
 203     FUNCDEF("constructor");
 204     LOG("+cleaner");
 205   }
 206
 207   virtual ~obsessive_compulsive() {
 208     FUNCDEF("destructor");
 209     LOG("~cleaner");
 210   }
 211
 212   DEFINE_CLASS_NAME("obsessive_compulsive");
 213
 214   void perform_activity(void *formal(data)) {
 215     FUNCDEF("perform_activity");
 216     while (!should_stop()) {
 217       // make sure there's nothing rotting too long.
 218       binger.clean_out_deadwood(DATA_DECAY_TIME);
 219       // snooze.
 220       int sleepy_time = randomizer().inclusive(MIN_TIDIER_THREAD_PAUSE,
 221           MAX_TIDIER_THREAD_PAUSE);
 222       time_control::sleep_ms(sleepy_time);
 223     }
 224   }
 225 };
 226
 227 // this thread will destroy all data in the bins while cleaning furiously.
 228 class monk_the_detective : public ethread
 229 {
 230 public:
 231   monk_the_detective() : ethread(0) {
 232     FUNCDEF("constructor");
 233     LOG("+monk");
 234   }
 235
 236   virtual ~monk_the_detective() {
 237     FUNCDEF("destructor");
 238     LOG("~monk");
 239   }
 240
 241   DEFINE_CLASS_NAME("monk_the_detective");
 242
 243   void perform_activity(void *formal(data)) {
 244     FUNCDEF("perform_activity");
 245     while (!should_stop()) {
 246       // one activation of monk has devastating consequences.  we empty out
 247       // the data one item at a time until we see no data at all.  after
 248       // cleaning each item, we ensure that the deadwood is cleaned out.
 249       binger._ent_lock->lock();
 250 LOG(a_sprintf("monk sees %d items.", binger.items_held()));
 251       while (binger.items_held()) {
 252         // grab one instance of any item in the bin.
 253         octopus_request_id id;
 254         infoton *found = binger.acquire_for_any(id);
 255         WHACK(found);
 256         // also clean out things a lot faster than normal.
 257         binger.clean_out_deadwood(MONKS_CLEANING_TIME);
 258       }
 259       binger._ent_lock->unlock();
 260 LOG(a_sprintf("after a little cleaning, monk sees %d items.", binger.items_held()));
 261       // snooze.
 262       int sleepy_time = randomizer().inclusive(MIN_MONK_THREAD_PAUSE,
 263           MAX_MONK_THREAD_PAUSE);
 264       time_control::sleep_ms(sleepy_time);
 265     }
 266   }
 267 };
 268
 269 //////////////
 270
 271 class test_entity_data_bin_threaded : public application_shell
 272 {
 273 public:
 274   test_entity_data_bin_threaded() : application_shell(class_name()) {}
 275
 276   DEFINE_CLASS_NAME("test_entity_data_bin_threaded");
 277
 278   int execute();
 279 };
 280
 281 int test_entity_data_bin_threaded::execute()
 282 {
 283   FUNCDEF("execute");
 284
 285   amorph<ethread> thread_list;
 286
 287   for (int i = 0; i < DEFAULT_THREADS; i++) {
 288     ethread *t = NIL;
 289     if (i == DEFAULT_THREADS - 1) {
 290       // last item gets special treatment; we reserve this space for monk.
 291       t = new monk_the_detective;
 292     } else if (i % 3 == 0) {
 293       t = new ballot_box_stuffer;
 294     } else if (i % 3 == 1) {
 295       t = new vote_destroyer;
 296     } else {  // i % 3 must = 2.
 297       t = new obsessive_compulsive;
 298     }
 299     thread_list.append(t);
 300     ethread *q = thread_list[thread_list.elements() - 1];
 301     if (q != t)
 302       deadly_error(class_name(), func, "amorph has incorrect pointer!");
 303     // start the thread we added.
 304     thread_list[thread_list.elements() - 1]->start(NIL);
 305   }
 306
 307   time_stamp when_to_leave(DEFAULT_RUN_TIME);
 308   while (when_to_leave > time_stamp()) {
 309     time_control::sleep_ms(100);
 310   }
 311
 312 //  LOG("now cancelling all threads....");
 313
 314 //  for (int j = 0; j < thread_list.elements(); j++) thread_list[j]->cancel();
 315
 316 //  LOG("now stopping all threads....");
 317
 318 //  for (int k = 0; k < thread_list.elements(); k++) thread_list[k]->stop();
 319
 320 //  LOG("resetting thread list....");
 321
 322   thread_list.reset();  // should whack all threads.
 323
 324   LOG("done exiting from all threads....");
 325
 326 //report the results:
 327 // how many objects created.
 328 // how many got destroyed.
 329 // how many evaporated due to timeout.
 330
 331
 332   guards::alert_message("t_bin_threaded:: works for all functions tested.");
 333   return 0;
 334 }
 335
 336 HOOPLE_MAIN(test_entity_data_bin_threaded, )
 337