text: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  #ifdef WIN32
  // ace.dll was built with FD_SETSIZE of 1024, so ensure it stays that way.
  #undef FD_SETSIZE
  #define FD_SETSIZE 1024
  #if WINVER == 0x0500
  #undef _WINSOCKAPI_
  #define NOMINMAX
  #include <WinSock2.h>
  #endif
  #endif
  
  #ifdef USE_SMARTHEAP
  #include <smrtheap.h>
  #endif
  
  #include "TimeBomb.hpp"
  
  #include <ace/ACE.h>
  
  #include <typeinfo>
  
  #include <string>
  #include <list>
  #include <map>
  
  // SW: Switching to framework BB on linux also since it is much faster.
  #ifndef _WIN32
  // On solaris, when ACE_Naming_Context maps file to memory using fixed mode, it
  // interfere with malloc/brk system calls later cause failure. For now, we use
  // the Black Board from the regression test framework. When the ACE problem is
  // fixed in a new release we'll go back to original code by undefining
  // SOLARIS_USE_BB
  #define SOLARIS_USE_BB 1
  #endif
  
  #define VALUE_MAX 128
  
  #ifdef SOLARIS_USE_BB
  #include "BBNamingContext.hpp"
  using apache::geode::client::testframework::BBNamingContextClient;
  using apache::geode::client::testframework::BBNamingContextServer;
  #else
  #include <ace/Naming_Context.h>
  #endif
  
  #include <ace/Guard_T.h>
  #include <ace/Get_Opt.h>
  #include <ace/Time_Value.h>
  #include <ace/SV_Semaphore_Complex.h>
  
  #include "fw_spawn.hpp"
  #include "fwklib/FwkException.hpp"
  
  #define __DUNIT_NO_MAIN__
  #include "fw_dunit.hpp"
  
  ACE_TCHAR *g_programName = nullptr;
  uint32_t g_coordinatorPid = 0;
  
  ClientCleanup gClientCleanup;
  
  namespace dunit {
  
  void HostWaitForDebugger() {
    int done = 0;
    LOG("host wait for debugger.");
    while (!done) {
      sleep(1);
    }
  }
  
  void setupCRTOutput() {
  #ifdef _WIN32
  #ifdef DEBUG
    int reportMode = _CRTDBG_MODE_FILE | _CRTDBG_MODE_WNDW;
    _CrtSetReportMode(_CRT_ASSERT, reportMode);
    _CrtSetReportFile(_CRT_ASSERT, _CRTDBG_FILE_STDERR);
    _CrtSetReportMode(_CRT_ERROR, reportMode);
    _CrtSetReportFile(_CRT_ERROR, _CRTDBG_FILE_STDERR);
    _CrtSetReportMode(_CRT_WARN, reportMode);
    _CrtSetReportFile(_CRT_WARN, _CRTDBG_FILE_STDERR);
    SetErrorMode(SEM_FAILCRITICALERRORS);
  #endif
  #endif
  }
  
  void getTimeStr(char *bufPtr, size_t sizeOfBuf) {
    ACE_TCHAR timestamp[64] = {0};  // only 35 needed here
    ACE::timestamp(timestamp, sizeof timestamp);
    // timestamp is like "Tue May 17 2005 12:54:22.546780"
    // for our purpose we just want "12:54:22.546780"
    strncpy(bufPtr, &timestamp[16], sizeOfBuf);
  }
  
  // some common values..
  #define WORKER_STATE_READY 1
  #define WORKER_STATE_DONE 2
  #define WORKER_STATE_TASK_ACTIVE 3
  #define WORKER_STATE_TASK_COMPLETE 4
  #define WORKER_STATE_SCHEDULED 5
  
  void log(std::string s, int lineno, const char *filename);
  
  /** Naming service for sharing data between processes. */
  class NamingContextImpl : virtual public NamingContext {
   private:
  #ifdef SOLARIS_USE_BB
    BBNamingContextClient
  #else
    ACE_Naming_Context
  #endif
        m_context;
  
    void millisleep(int msec) {
      ACE_Time_Value sleepTime;
      sleepTime.msec(msec);
      ACE_OS::sleep(sleepTime);
    }
  
    int checkResult(int result, const char *func) {
      if (result == -1) {
        LOGCOORDINATOR("NamingCtx operation failed for:");
        LOGCOORDINATOR(func);
        LOGCOORDINATOR("Dump follows:");
        dump();
        throw - 1;
      }
      return result;
    }
  
   public:
    NamingContextImpl() : m_context() {
      open();
      LOGCOORDINATOR("Naming context ready.");
    }
  
    ~NamingContextImpl() noexcept override {
      m_context.close();
      ACE_OS::unlink(ACE_OS::getenv("TESTNAME"));
    }
  
    /**
     * Share a string value, return -1 if there is a failure to store value,
     * otherwise returns 0.
     */
    int rebind(const char *key, const char *value) override {
      int res = -1;
      int attempts = 10;
      while ((res = m_context.rebind(key, value, const_cast<char *>(""))) == -1 &&
             attempts--) {
        millisleep(10);
      }
      return checkResult(res, "rebind");
    }
  
    /**
     * Share an int value, return -1 if there is a failure to store value,
     * otherwise returns 0.
     */
    int rebind(const char *key, int value) override {
      return rebind(key, std::to_string(value).c_str());
    }
  
    /**
     * retreive a value by key, storing the result in the users buf. If the key
     * is not found, the buf will contain the empty string "".
     */
    void getValue(const char *key, char *buf, size_t sizeOfBuf) override {
  #ifdef SOLARIS_USE_BB
      char value[VALUE_MAX] = {0};
      char type[VALUE_MAX] = {0};
  #else
      char *value = nullptr;
      char *type = nullptr;
  #endif
  
      int res = -1;
      // we should not increase attempts to avoid increasing test run times.
      int attempts = 3;
      while ((res = m_context.resolve(key, value, type)) != 0 && attempts--) {
        // we should not increase sleep to avoid increasing test run times.
        millisleep(5);
      }
  
      if (res != 0) {
        strncpy(buf, "", sizeOfBuf);
        return;
      }
      ACE_OS::strncpy(buf, value, sizeOfBuf);
    }
  
    /**
     * return the value by key, as an int using the string to int conversion
     * rules of atoi.
     */
    int getIntValue(const char *key) override {
      char value[VALUE_MAX] = {0};
      getValue(key, value, sizeof(value));
      if (ACE_OS::strcmp(value, "") == 0) return 0;
      return ACE_OS::atoi(value);
    }
  
    void open() {
  #ifdef SOLARIS_USE_BB
      m_context.open();
  #else
      ACE_Name_Options *name_options = m_context.name_options();
      name_options->process_name(getContextName().c_str());
      name_options->namespace_dir(".");
      name_options->context(ACE_Naming_Context::PROC_LOCAL);
      name_options->database(ACE_OS::getenv("TESTNAME"));
      checkResult(m_context.open(name_options->context(), 0), "open");
  #endif
      LOGCOORDINATOR("Naming context opened.");
    }
  
    std::string getContextName() {
      char buf[1024] = {0};
      ACE_OS::sprintf(buf, "dunit.context.%s%d", ACE::basename(g_programName),
                      g_coordinatorPid);
      std::string b_str(buf);
      return b_str;
    }
  
    std::string getMutexName() {
      char buf[1024] = {0};
      ACE_OS::sprintf(buf, "dunit.mutex.%s%d", ACE::basename(g_programName),
                      g_coordinatorPid);
      std::string b_str(buf);
      return b_str;
    }
  
    /** print out all the entries' keys and values in the naming context. */
    void dump() override {
  #ifdef SOLARIS_USE_BB
      m_context.dump();
  #else
      ACE_BINDING_SET set;
      if (this->m_context.list_name_entries(set, "") != 0) {
        char buf[1000] = {0};
        ACE_OS::sprintf(buf, "There is nothing in the naming context.");
        LOGCOORDINATOR(buf);
      } else {
        ACE_BINDING_ITERATOR set_iterator(set);
        for (ACE_Name_Binding *entry = 0; set_iterator.next(entry) != 0;
             set_iterator.advance()) {
          ACE_Name_Binding binding(*entry);
          char buf[1000] = {0};
          ACE_OS::sprintf(buf, "%s => %s", binding.name_.char_rep(),
                          binding.value_.char_rep());
          LOGCOORDINATOR(buf);
        }
      }
  #endif
    }
  
    void resetContext() {
      char buf[30] = {0};
      sprintf(buf, "%d", ACE_OS::getpid());
  
      int res1 = -1;
      int attempts1 = 10;
      while ((res1 = m_context.rebind("Driver", buf)) == -1 && attempts1--) {
        millisleep(10);
      }
      checkResult(res1, "rebind1");
  
      int res2 = -1;
      int attempts2 = 10;
      while ((res2 = m_context.rebind("WorkerId", "0")) == -1 && attempts2--) {
        millisleep(10);
      }
      checkResult(res2, "rebind2");
  
      LOGCOORDINATOR("Naming context reset.");
    }
  };
  
  /** uniquely represent each different worker. */
  class WorkerId {
   private:
    uint32_t m_id;
    static const char *m_idNames[];
  
   public:
    explicit WorkerId(uint32_t id) { m_id = id; }
  
    int getId() { return m_id; }
  
    const char *getIdName() { return m_idNames[m_id]; }
  
    /** return the system id for this process */
    int getSystem() { return 1; }
    /** return the process id for this system. */
    int getProcOnSys() { return ((m_id % 2) == 0) ? 2 : 1; }
  };
  
  const char *WorkerId::m_idNames[] = {"none", "s1p1", "s1p2", "s2p1", "s2p2"};
  
  /** method for letting Task discover its name through RTTI. */
  std::string Task::typeName() { return std::string(typeid(*this).name()); }
  
  typedef std::list<Task *> TaskList;
  
  /** contains a queue of Task* for each WorkerId. */
  class TaskQueues {
   private:
    std::map<int, TaskList> m_qmap;
    std::list<int> m_schedule;
  
    TaskQueues() : m_qmap(), m_schedule() {}
  
    void registerTask(WorkerId sId, Task *task) {
      m_qmap[sId.getId()].push_back(task);
      m_schedule.push_back(sId.getId());
    }
  
    Task *nextTask(WorkerId &sId) {
      TaskList *tasks = &(m_qmap[sId.getId()]);
      if (tasks->empty()) {
        return nullptr;
      }
      Task *task = tasks->front();
      if (task != nullptr) {
        char logmsg[1024] = {0};
        sprintf(logmsg, "received task: %s ", task->m_taskName.c_str());
        LOG(logmsg);
        tasks->pop_front();
      }
      return task;
    }
  
    int nextWorkerId() {
      if (m_schedule.empty()) {
        return 0;
      }
      int sId = m_schedule.front();
      char logmsg[1024] = {0};
      sprintf(logmsg, "Next worker id is : %d", sId);
      LOGCOORDINATOR(logmsg);
      m_schedule.pop_front();
      return sId;
    }
  
    static TaskQueues *taskQueues;
  
   public:
    static void addTask(WorkerId sId, Task *task) {
      if (taskQueues == nullptr) {
        taskQueues = new TaskQueues();
      }
      taskQueues->registerTask(sId, task);
    }
  
    static int getWorkerId() {
      ASSERT(taskQueues != nullptr, "failure to initialize fw_dunit module.");
      return taskQueues->nextWorkerId();
    }
  
    static Task *getTask(WorkerId sId) {
      ASSERT(taskQueues != nullptr, "failure to initialize fw_dunit module.");
      return taskQueues->nextTask(sId);
    }
  };
  
  TaskQueues *TaskQueues::taskQueues = nullptr;
  
  /** register task with worker. */
  void Task::init(int sId) { init(sId, false); }
  
  void Task::init(int sId, bool isHeapAllocated) {
    m_isHeapAllocated = isHeapAllocated;
    m_id = sId;
    m_taskName = this->typeName();
    TaskQueues::addTask(WorkerId(sId), this);
  }
  
  /** main framework entry */
  class Dunit {
   private:
    NamingContextImpl m_globals;
    static Dunit *singleton;
    bool m_close_down;
  
    Dunit() : m_globals(), m_close_down(false) {}
  
    void resetContext() {
      m_close_down = true;
      m_globals.resetContext();
    }
  
   public:
    /** call this once just inside main... */
    static void init(bool initContext = false) {
      if (initContext) {
        ACE_OS::unlink("localnames");
        ACE_OS::unlink("name_space_localnames");
        ACE_OS::unlink("backing_store_localnames");
      }
      singleton = new Dunit();
      if (initContext) {
        singleton->resetContext();
      }
    }
  
    /** return the already initialized singleton Dunit instance. */
    static Dunit *getSingleton() {
      ASSERT(singleton != nullptr, "singleton not created yet.");
      return singleton;
    }
  
    /** delete the existing singleton */
    static void close() {
      Dunit *tmp = singleton;
      singleton = nullptr;
      delete tmp;
    }
  
    /** set the next worker id */
    void setNextWorker(WorkerId &sId) {
      m_globals.rebind("WorkerId", sId.getId());
    }
  
    /** get the next worker id */
    int getNextWorker() { return m_globals.getIntValue("WorkerId"); }
  
    /** return true if all workers are to terminate. */
    bool mustQuit() {
      return m_globals.getIntValue("TerminateAllWorkers") ? true : false;
    }
  
    /** signal all workers to terminate. */
    void setMustQuit() { m_globals.rebind("TerminateAllWorkers", 1); }
  
    /** signal to test driver that an error occurred. */
    void setFailed() { m_globals.rebind("Failure", 1); }
  
    bool getFailed() { return m_globals.getIntValue("Failure") ? true : false; }
  
    void setWorkerState(WorkerId sId, int state) {
      char key[100] = {0};
      ACE_OS::sprintf(key, "ReadyWorker%d", sId.getId());
      m_globals.rebind(key, state);
    }
  
    int getWorkerState(WorkerId sId) {
      char key[100] = {0};
      ACE_OS::sprintf(key, "ReadyWorker%d", sId.getId());
      return m_globals.getIntValue(key);
    }
  
    void setWorkerTimeout(WorkerId sId, int seconds) {
      char key[100] = {0};
      ACE_OS::sprintf(key, "TimeoutWorker%d", sId.getId());
      m_globals.rebind(key, seconds);
    }
  
    int getWorkerTimeout(WorkerId sId) {
      char key[100] = {0};
      ACE_OS::sprintf(key, "TimeoutWorker%d", sId.getId());
      return m_globals.getIntValue(key);
    }
  
    /** return the NamingContext for global (amongst all processes) values. */
    NamingContext *globals() { return &m_globals; }
  
    ~Dunit() {}
  };
  
  #define DUNIT dunit::Dunit::getSingleton()
  
  Dunit *Dunit::singleton = nullptr;
  
  void Task::setTimeout(int seconds) {
    if (seconds > 0) {
      DUNIT->setWorkerTimeout(WorkerId(m_id), seconds);
    } else {
      DUNIT->setWorkerTimeout(WorkerId(m_id), TASK_TIMEOUT);
    }
  }
  
  class TestProcess : virtual public dunit::Manager {
   private:
    WorkerId m_sId;
  
   public:
    TestProcess(const ACE_TCHAR *cmdline, uint32_t id)
        : Manager(cmdline), m_sId(id) {}
  
    WorkerId &getWorkerId() { return m_sId; }
  
   protected:
   public:
    virtual ~TestProcess() {}
  };
  
  /**
   * Container of TestProcess(es) held in driver. each represents one of the
   * legal WorkerIds spawned when TestDriver is created.
   */
  class TestDriver {
   private:
    TestProcess *m_workers[4];
  #ifdef SOLARIS_USE_BB
    BBNamingContextServer *m_bbNamingContextServer;
  #endif
  
   public:
    TestDriver() {
  #ifdef SOLARIS_USE_BB
      m_bbNamingContextServer = new BBNamingContextServer();
      ACE_OS::sleep(5);
      fprintf(stdout, "Blackboard started\n");
      fflush(stdout);
  #endif
  
      dunit::Dunit::init(true);
      fprintf(stdout, "Coordinator starting workers.\n");
      for (uint32_t i = 1; i < 5; i++) {
        ACE_TCHAR cmdline[2048] = {0};
        char *profilerCmd = ACE_OS::getenv("PROFILERCMD");
        if (profilerCmd != nullptr && profilerCmd[0] != '$' &&
            profilerCmd[0] != '\0') {
          // replace %d's in profilerCmd with PID and worker ID
          char cmdbuf[2048] = {0};
          ACE_OS::sprintf(cmdbuf, profilerCmd, ACE_OS::gettimeofday().msec(),
                          g_coordinatorPid, i);
          ACE_OS::sprintf(cmdline, "%s %s -s%d -m%d", cmdbuf, g_programName, i,
                          g_coordinatorPid);
        } else {
          ACE_OS::sprintf(cmdline, "%s -s%d -m%d", g_programName, i,
                          g_coordinatorPid);
        }
        fprintf(stdout, "%s\n", cmdline);
        m_workers[i - 1] = new TestProcess(cmdline, i);
      }
      fflush(stdout);
      // start each of the workers...
      for (uint32_t j = 1; j < 5; j++) {
        m_workers[j - 1]->doWork();
        ACE_OS::sleep(2);  // do not increase this to avoid precheckin runs taking
                           // much longer.
      }
    }
  
    ~TestDriver() {
      // kill off any children that have not yet terminated.
      for (uint32_t i = 1; i < 5; i++) {
        if (m_workers[i - 1]->running() == 1) {
          delete m_workers[i - 1];  // worker destructor should terminate process.
        }
      }
      dunit::Dunit::close();
  #ifdef SOLARIS_USE_BB
      delete m_bbNamingContextServer;
      m_bbNamingContextServer = nullptr;
  #endif
    }
  
    int begin() {
      fprintf(stdout, "Coordinator started with pid %d\n", ACE_OS::getpid());
      fflush(stdout);
      waitForReady();
      // dispatch task...
  
      int nextWorker;
      while ((nextWorker = TaskQueues::getWorkerId()) != 0) {
        WorkerId sId(nextWorker);
        DUNIT->setWorkerState(sId, WORKER_STATE_SCHEDULED);
        fprintf(stdout, "Set next process to %s\n", sId.getIdName());
        fflush(stdout);
        DUNIT->setNextWorker(sId);
        waitForCompletion(sId);
        // check special conditions.
        if (DUNIT->getFailed()) {
          DUNIT->setMustQuit();
          waitForDone();
          return 1;
        }
      }
  
      // end all work..
      DUNIT->setMustQuit();
      waitForDone();
      return 0;
    }
  
    /** wait for an individual worker to finish a task. */
    void waitForCompletion(WorkerId &sId) {
      int secs = DUNIT->getWorkerTimeout(sId);
      DUNIT->setWorkerTimeout(sId, TASK_TIMEOUT);
      if (secs <= 0) secs = TASK_TIMEOUT;
      fprintf(stdout, "Waiting %d seconds for %s to finish task.\n", secs,
              sId.getIdName());
      fflush(stdout);
      ACE_Time_Value end = ACE_OS::gettimeofday();
      ACE_Time_Value offset(secs, 0);
      end += offset;
      while (DUNIT->getWorkerState(sId) != WORKER_STATE_TASK_COMPLETE) {
        // sleep a bit..
        if (DUNIT->getFailed()) return;
        ACE_Time_Value sleepTime;
        sleepTime.msec(100);
        ACE_OS::sleep(sleepTime);
        checkWorkerDeath();
        ACE_Time_Value now = ACE_OS::gettimeofday();
        if (now >= end) {
          handleTimeout(sId);
          break;
        }
      }
    }
  
    void handleTimeout() {
      fprintf(stdout, "Error: Timed out waiting for all workers to be ready.\n");
      fflush(stdout);
      DUNIT->setMustQuit();
      DUNIT->setFailed();
    }
  
    void handleTimeout(WorkerId &sId) {
      fprintf(stdout, "Error: Timed out waiting for %s to finish task.\n",
              sId.getIdName());
      fflush(stdout);
      DUNIT->setMustQuit();
      DUNIT->setFailed();
    }
  
    /** wait for all workers to be done initializing. */
    void waitForReady() {
      fprintf(stdout, "Waiting %d seconds for all workers to be ready.\n",
              TASK_TIMEOUT);
      fflush(stdout);
      ACE_Time_Value end = ACE_OS::gettimeofday();
      ACE_Time_Value offset(TASK_TIMEOUT, 0);
      end += offset;
      uint32_t readyCount = 0;
      while (readyCount < 4) {
        fprintf(stdout, "Ready Count: %d\n", readyCount);
        fflush(stdout);
        if (DUNIT->getFailed()) return;
        // sleep a bit..
        ACE_Time_Value sleepTime(1);
        //      sleepTime.msec( 10 );
        ACE_OS::sleep(sleepTime);
        readyCount = 0;
        for (uint32_t i = 1; i < 5; i++) {
          int state = DUNIT->getWorkerState(WorkerId(i));
          if (state == WORKER_STATE_READY) {
            readyCount++;
          }
        }
        checkWorkerDeath();
        ACE_Time_Value now = ACE_OS::gettimeofday();
        if (now >= end) {
          handleTimeout();
          break;
        }
      }
    }
  
    /** wait for all workers to be destroyed. */
    void waitForDone() {
      fprintf(stdout, "Waiting %d seconds for all workers to complete.\n",
              TASK_TIMEOUT);
      fflush(stdout);
      ACE_Time_Value end = ACE_OS::gettimeofday();
      ACE_Time_Value offset(TASK_TIMEOUT, 0);
      end += offset;
      uint32_t doneCount = 0;
      while (doneCount < 4) {
        // if ( DUNIT->getFailed() ) return;
        // sleep a bit..
        ACE_Time_Value sleepTime;
        sleepTime.msec(100);
        ACE_OS::sleep(sleepTime);
        doneCount = 0;
        for (uint32_t i = 1; i < 5; i++) {
          int state = DUNIT->getWorkerState(WorkerId(i));
          if (state == WORKER_STATE_DONE) {
            doneCount++;
          }
        }
        ACE_Time_Value now = ACE_OS::gettimeofday();
        if (now >= end) {
          handleTimeout();
          break;
        }
      }
    }
  
    /** test to see that all the worker processes are still around, or throw
        a TestException so the driver doesn't get hung. */
    void checkWorkerDeath() {
      for (uint32_t i = 0; i < 4; i++) {
        if (!m_workers[i]->running()) {
          char msg[1000] = {0};
          sprintf(msg, "Error: Worker %s terminated prematurely.",
                  m_workers[i]->getWorkerId().getIdName());
          LOG(msg);
          DUNIT->setFailed();
          DUNIT->setMustQuit();
          FAIL(msg);
        }
      }
    }
  };
  
  class TestWorker {
   private:
    WorkerId m_sId;
  
   public:
    static WorkerId *procWorkerId;
  
    explicit TestWorker(int id) : m_sId(id) {
      procWorkerId = new WorkerId(id);
      dunit::Dunit::init();
      DUNIT->setWorkerState(m_sId, WORKER_STATE_READY);
    }
  
    ~TestWorker() {
      DUNIT->setWorkerState(m_sId, WORKER_STATE_DONE);
      dunit::Dunit::close();
    }
  
    void begin() {
      fprintf(stdout, "Worker %s started with pid %d\n", m_sId.getIdName(),
              ACE_OS::getpid());
      fflush(stdout);
      WorkerId workerZero(0);
  
      // consume tasks of this workers queue, only when it is his turn..
      while (!DUNIT->mustQuit()) {
        if (DUNIT->getNextWorker() == m_sId.getId()) {
          // set next worker to zero so I don't accidently run twice.
          DUNIT->setNextWorker(workerZero);
          // do next task...
          Task *task = TaskQueues::getTask(m_sId);
          // perform task.
          if (task != nullptr) {
            DUNIT->setWorkerState(m_sId, WORKER_STATE_TASK_ACTIVE);
            try {
              task->doTask();
              if (task->m_isHeapAllocated) {
                delete task;
              }
              fflush(stdout);
              DUNIT->setWorkerState(m_sId, WORKER_STATE_TASK_COMPLETE);
            } catch (TestException te) {
              if (task->m_isHeapAllocated) {
                delete task;
              }
              te.print();
              handleError();
              return;
            } catch (...) {
              if (task->m_isHeapAllocated) {
                delete task;
              }
              LOG("Unhandled exception, terminating.");
              handleError();
              return;
            }
          }
        }
        ACE_Time_Value sleepTime;
        sleepTime.msec(100);
        ACE_OS::sleep(sleepTime);
      }
    }
  
    void handleError() {
      DUNIT->setFailed();
      DUNIT->setMustQuit();
      DUNIT->setWorkerState(m_sId, WORKER_STATE_TASK_COMPLETE);
    }
  };
  
  WorkerId *TestWorker::procWorkerId = nullptr;
  
  void sleep(int millis) {
    if (millis == 0) {
      ACE_OS::thr_yield();
    } else {
      ACE_Time_Value sleepTime;
      sleepTime.msec(millis);
      ACE_OS::sleep(sleepTime);
    }
  }
  
  void logCoordinator(std::string s, int lineno, const char * /*filename*/) {
    char buf[128] = {0};
    dunit::getTimeStr(buf, sizeof(buf));
  
    fprintf(stdout, "[TEST coordinator:pid(%d)] %s at line: %d\n",
            ACE_OS::getpid(), s.c_str(), lineno);
    fflush(stdout);
  }
  
  // log a message and print the worker id as well.. used by fw_helper with no
  // worker id.
  void log(std::string s, int lineno, const char * /*filename*/, int /*id*/) {
    char buf[128] = {0};
    dunit::getTimeStr(buf, sizeof(buf));
  
    fprintf(stdout, "[TEST 0:pid(%d)] %s at line: %d\n", ACE_OS::getpid(),
            s.c_str(), lineno);
    fflush(stdout);
  }
  
  // log a message and print the worker id as well..
  void log(std::string s, int lineno, const char * /*filename*/) {
    char buf[128] = {0};
    dunit::getTimeStr(buf, sizeof(buf));
  
    fprintf(stdout, "[TEST %s %s:pid(%d)] %s at line: %d\n", buf,
            (dunit::TestWorker::procWorkerId
                 ? dunit::TestWorker::procWorkerId->getIdName()
                 : "coordinator"),
            ACE_OS::getpid(), s.c_str(), lineno);
    fflush(stdout);
  }
  
  void cleanup() { gClientCleanup.callClientCleanup(); }
  
  int dmain(int argc, ACE_TCHAR *argv[]) {
  #ifdef USE_SMARTHEAP
    MemRegisterTask();
  #endif
    setupCRTOutput();
    TimeBomb tb(&cleanup);
    // tb->arm(); // leak this on purpose.
    try {
      g_programName = new ACE_TCHAR[2048];
      ACE_OS::strncpy(g_programName, argv[0], 2048);
  
      const ACE_TCHAR options[] = ACE_TEXT("s:m:");
      ACE_Get_Opt cmd_opts(argc, argv, options);
  
      int result = 0;
  
      int workerId = 0;
      int option = 0;
      while ((option = cmd_opts()) != EOF) {
        switch (option) {
          case 's':
            workerId = ACE_OS::atoi(cmd_opts.opt_arg());
            fprintf(stdout, "Using process id: %d\n", workerId);
            fflush(stdout);
            break;
          case 'm':
            g_coordinatorPid = ACE_OS::atoi(cmd_opts.opt_arg());
            fprintf(stdout, "Using coordinator id: %d\n", g_coordinatorPid);
            fflush(stdout);
            break;
          default:
            fprintf(stdout, "ignoring option: %s  with value %s\n",
                    cmd_opts.last_option(), cmd_opts.opt_arg());
            fflush(stdout);
        }
      }
  
      //  perf::NamingServiceThread nsvc( 12045 );
      //  nsvc.activate( THR_NEW_LWP | THR_DETACHED | THR_DAEMON, 1 );
      //  dunit::Dunit::init( true );
      //
      //  for ( int i = cmd_opts.opt_ind(); i < argc; i++ ) {
      //    char buf[1024], * name, * value;
      //    strcpy( buf, argv[i] );
      //    name = &buf[0];
      //    value = strchr( name, '=' );
      //    if ( value != 0 ) {
      //      *value = '\0';
      //      value++;
      //      // add to context
      //      dunit::globals()->rebind( name, value );
      //    }
      //  }
  
      // record the coordinator pid if it wasn't passed to us on the command line.
      // the TestDriver will pass this to the child processes.
      // currently this is used for giving a unique per run id to shared
      // resources.
      if (g_coordinatorPid == 0) {
        g_coordinatorPid = ACE_OS::getpid();
      }
  
      if (workerId > 0) {
        dunit::TestWorker worker(workerId);
        worker.begin();
      } else {
        dunit::TestDriver tdriver;
        result = tdriver.begin();
        if (result == 0) {
          printf("#### All Tasks completed successfully. ####\n");
        } else {
          printf("#### FAILED. ####\n");
        }
  
        fflush(stdout);
      }
      printf("final worker id %d, result %d\n", workerId, result);
      printf("before calling cleanup %d \n", workerId);
      gClientCleanup.callClientCleanup();
      printf("after calling cleanup\n");
      return result;
  
    } catch (dunit::TestException &te) {
      te.print();
    } catch (apache::geode::client::testframework::FwkException &fe) {
      printf("Exception: %s\n", fe.what());
      fflush(stdout);
    } catch (...) {
      printf("Exception: unhandled/unidentified exception reached main.\n");
      fflush(stdout);
      // return 1;
    }
    gClientCleanup.callClientCleanup();
    return 1;
  }
  
  /** entry point for test code modules to access the naming service. */
  NamingContext *globals() { return DUNIT->globals(); }
  
  }  // namespace dunit
  
  namespace perf {
  
  TimeStamp::TimeStamp(int64_t msec) : m_msec(msec) {}
  
  TimeStamp::TimeStamp() {
    ACE_Time_Value tmp = ACE_OS::gettimeofday();
    m_msec = tmp.msec();
  }
  
  TimeStamp::TimeStamp(const TimeStamp &other) : m_msec(other.m_msec) {}
  
  TimeStamp &TimeStamp::operator=(const TimeStamp &other) {
    m_msec = other.m_msec;
    return *this;
  }
  
  TimeStamp::~TimeStamp() {}
  
  int64_t TimeStamp::msec() const { return m_msec; }
  
  void TimeStamp::msec(int64_t t) { m_msec = t; }
  
  Record::Record(std::string testName, int64_t ops, const TimeStamp &start,
                 const TimeStamp &stop)
      : m_testName(testName),
        m_operations(ops),
        m_startTime(start),
        m_stopTime(stop) {}
  
  Record::Record()
      : m_testName(""), m_operations(0), m_startTime(0), m_stopTime(0) {}
  
  Record::Record(const Record &other)
      : m_testName(other.m_testName),
        m_operations(other.m_operations),
        m_startTime(other.m_startTime),
        m_stopTime(other.m_stopTime) {}
  
  Record &Record::operator=(const Record &other) {
    m_testName = other.m_testName;
    m_operations = other.m_operations;
    m_startTime = other.m_startTime;
    m_stopTime = other.m_stopTime;
    return *this;
  }
  
  void Record::write(apache::geode::client::DataOutput &output) {
    output.writeString(m_testName);
    output.writeInt(m_operations);
    output.writeInt(m_startTime.msec());
    output.writeInt(m_stopTime.msec());
  }
  
  void Record::read(apache::geode::client::DataInput &input) {
    m_testName = input.readString();
    m_operations = input.readInt64();
    m_startTime.msec(input.readInt64());
    m_stopTime.msec(input.readInt64());
  }
  
  Record::~Record() {}
  
  int Record::elapsed() {
    return static_cast<int>(m_stopTime.msec() - m_startTime.msec());
  }
  
  int Record::perSec() {
    return static_cast<int>(((static_cast<double>(1000) * m_operations) /
                             static_cast<double>(elapsed())) +
                            0.5);
  }
  
  std::string Record::asString() {
    std::string tmp = m_testName;
    char *buf = new char[1000];
    sprintf(buf, " -- %d ops/sec, ", perSec());
    tmp += buf;
    sprintf(buf, "%d ops, ", static_cast<int>(m_operations));
    tmp += buf;
    sprintf(buf, "%d millis", elapsed());
    tmp += buf;
    return tmp;
  }
  
  PerfSuite::PerfSuite(const char *suiteName) : m_suiteName(suiteName) {}
  
  void PerfSuite::addRecord(std::string testName, int64_t ops,
                            const TimeStamp &start, const TimeStamp &stop) {
    Record tmp(testName, ops, start, stop);
    m_records[testName] = tmp;
    fprintf(stdout, "[PerfSuite] %s\n", tmp.asString().c_str());
    fflush(stdout);
  }
  
  /** create a file in cwd, named "<suite>_results.<host>" */
  void PerfSuite::save() {
    /* Currently having trouble with windows... not useful until the compare
       function is written anyway...
  
    apache::geode::client::DataOutput output;
    output.writeASCII( m_suiteName.c_str(), m_suiteName.length() );
  
    char hname[100];
    ACE_OS::hostname( hname, 100 );
    std::string fname = m_suiteName + "_results." + hname;
  
    output.writeASCII( hname );
  
    for( RecordMap::iterator iter = m_records.begin(); iter != m_records.end();
    iter++ ) {
      Record record = (*iter).second;
      record.write( output );
    }
    fprintf( stdout, "[PerfSuite] finished serializing results.\n" );
    fflush( stdout );
  
    fprintf( stdout, "[PerfSuite] writing results to %s\n", fname.c_str() );
    FILE* of = ACE_OS::fopen( fname.c_str(), "a+" );
    if ( of == 0 ) {
      FAIL( "failed to open result file handle for perfSuite." );
    }
    LOG( "opened perf output file for a+" );
    uint32_t len = 0;
    char* buf = (char*) output.getBuffer( &len );
    LOG( "got buffer." );
    ACE_OS::fwrite( buf, len, 1, of );
    LOG( "wrote buffer" );
    ACE_OS::fflush( of );
    LOG( "flushed of" );
    ACE_OS::fclose( of );
    LOG( "closed of" );
    fprintf( stdout, "[PerfSuite] finished saving results file %s\n",
    fname.c_str() );
    fflush( stdout );
    */
  }
  
  /** load data saved in $ENV{'baselines'} named "<suite>_baseline.<host>" */
  void PerfSuite::compare() {
    char hname[100] = {0};
    ACE_OS::hostname(hname, 100);
    std::string fname = m_suiteName + "_baseline." + hname;
  }
  
  ThreadLauncher::ThreadLauncher(int thrCount, Thread &thr)
      : m_thrCount(thrCount),
        m_initSemaphore((-1 * thrCount) + 1),
        m_startSemaphore(0),
        m_stopSemaphore((-1 * thrCount) + 1),
        m_cleanSemaphore(0),
        m_termSemaphore((-1 * thrCount) + 1),
        m_startTime(nullptr),
        m_stopTime(nullptr),
        m_threadDef(thr) {
    m_threadDef.init(this);
  }
  
  void ThreadLauncher::go() {
  #ifdef WIN32
    int thrAttrs = THR_NEW_LWP | THR_JOINABLE;
  #else
    int thrAttrs = THR_NEW_LWP | THR_JOINABLE | THR_INHERIT_SCHED;
  #endif
    int res = m_threadDef.activate(thrAttrs, m_thrCount);
    ASSERT(res == 0, "Failed to start threads properly");
    m_initSemaphore.acquire();
    LOG("[ThreadLauncher] all threads ready.");
    m_startTime = new TimeStamp();
    m_startSemaphore.release(m_thrCount);
    m_stopSemaphore.acquire();
    m_stopTime = new TimeStamp();
    m_cleanSemaphore.release(m_thrCount);
    m_termSemaphore.acquire();
    LOG("[ThreadLauncher] joining threads.");
    m_threadDef.wait();
    LOG("[ThreadLauncher] all threads stopped.");
  }
  
  ThreadLauncher::~ThreadLauncher() {
    if (m_startTime) {
      delete m_startTime;
    }
    if (m_stopTime) {
      delete m_stopTime;
    }
  }
  
  Thread::Thread() : ACE_Task_Base(), m_launcher(nullptr), m_used(false) {}
  
  Thread::~Thread() {}
  
  int Thread::svc() {
    m_used = true;
    int res = 0;
    try {
      setup();  // do per thread setup
    } catch (...) {
      LOG("[Thread] unknown exception thrown in setup().");
      res = 1;
    }
    m_launcher->initSemaphore().release();
    m_launcher->startSemaphore().acquire();
    try {
      perftask();  // do measured iterations
    } catch (...) {
      LOG("[Thread] unknown exception thrown in perftask().");
      res = 2;
    }
    m_launcher->stopSemaphore().release();
    m_launcher->cleanSemaphore().acquire();
    try {
      cleanup();  // cleanup after thread.
    } catch (...) {
      LOG("[Thread] unknown exception thrown in cleanup()");
      res = 3;
    }
    m_launcher->termSemaphore().release();
    return res;
  }
  
  Semaphore::Semaphore(int count) : m_mutex(), m_cond(m_mutex), m_count(count) {}
  
  Semaphore::~Semaphore() {}
  
  void Semaphore::acquire(int t) {
    ACE_Guard<ACE_Thread_Mutex> _guard(m_mutex);
  
    while (m_count < t) {
      m_cond.wait();
    }
    m_count -= t;
  }
  
  void Semaphore::release(int t) {
    ACE_Guard<ACE_Thread_Mutex> _guard(m_mutex);
  
    m_count += t;
    if (m_count > 0) {
      m_cond.broadcast();
    }
  }
  
  }  // namespace perf
  