// Perform a torture test of multiple ACE_Reactors and ACE_Tasks in
// the same process...  Thanks to Detlef Becker for contributing this.

#include "ace/Reactor.h"
#include "ace/Service_Config.h"
#include "ace/Task.h"
#include "ace/Atomic_Op.h"



#if defined (ACE_HAS_THREADS)

#include "ace/Recursive_Thread_Mutex.h"

static const int NUM_INVOCATIONS = 10;
static const int MAX_TASKS = 20;

class Test_Task : public ACE_Task<ACE_MT_SYNCH>
{
public:
  Test_Task (void);
  ~Test_Task (void);

  //FUZZ: disable check_for_lack_ACE_OS
  virtual int open (void *args = 0);
  virtual int close (u_long flags = 0);
  //FUZZ: enable check_for_lack_ACE_OS

  virtual int svc ();

  virtual int handle_input (ACE_HANDLE handle);
  virtual int handle_close (ACE_HANDLE fd,
                            ACE_Reactor_Mask close_mask);

private:
  int handled_;

  static int task_count_;
};

int Test_Task::task_count_ = 0;

static ACE_Atomic_Op<ACE_Thread_Mutex, int> done_count = MAX_TASKS * 2;

static ACE_Recursive_Thread_Mutex reclock_;

Test_Task::Test_Task (void)
  : handled_ (0)
{
  ACE_GUARD (ACE_Recursive_Thread_Mutex, ace_mon, reclock_);

  Test_Task::task_count_++;
  ACE_DEBUG ((LM_DEBUG,
              "(%t) TT+ Test_Task::task_count_ = %d\n",
              Test_Task::task_count_));
}

Test_Task::~Test_Task (void)
{
  ACE_GUARD (ACE_Recursive_Thread_Mutex, ace_mon, reclock_);

  ACE_DEBUG ((LM_DEBUG,
              "(%t) TT- Test_Task::task_count_ = %d\n",
              Test_Task::task_count_));
}

int
Test_Task::open (void *args)
{
  this->reactor ((ACE_Reactor *) args);
  return this->activate (THR_NEW_LWP);
}

int
Test_Task::close (u_long)
{
  ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, ace_mon, reclock_, -1);

  Test_Task::task_count_--;
  ACE_DEBUG ((LM_DEBUG,
              "(%t) close Test_Task::task_count_ = %d\n",
              Test_Task::task_count_));
  return 0;
}

int
Test_Task::svc (void)
{
  for (int i = 0; i < NUM_INVOCATIONS; i++)
    {
      ACE_OS::thr_yield ();

      // ACE_DEBUG ((LM_DEBUG, "(%t) calling notify %d\n", i));

      if (this->reactor ()->notify (this, ACE_Event_Handler::READ_MASK) == -1)
        ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "notify"), -1);

      // ACE_DEBUG ((LM_DEBUG, "(%t) leaving notify %d\n", i));
    }

  return 0;
}

int
Test_Task::handle_close (ACE_HANDLE,
                         ACE_Reactor_Mask)
{
  ACE_DEBUG ((LM_DEBUG, "(%t) handle_close\n"));
  return 0;
}

int
Test_Task::handle_input (ACE_HANDLE)
{
  ACE_DEBUG ((LM_DEBUG, "(%t) handle_input\n"));

  this->handled_++;

  if (this->handled_ == NUM_INVOCATIONS)
    {
      done_count--;
      ACE_DEBUG ((LM_DEBUG,
                  "(%t) handle_input, handled_ = %d, done_count = %d\n",
                  this->handled_, done_count.value ()));
    }

  ACE_OS::thr_yield ();
  return -1;
}

static void *
worker (void *args)
{
  ACE_Reactor *reactor = (ACE_Reactor *) args;

  reactor->owner (ACE_Thread::self ());

  ACE_Time_Value timeout (4);

  for (;;)
    {
      //ACE_DEBUG ((LM_DEBUG, "(%t) calling handle_events\n"));

      switch (reactor->handle_events (timeout))
        {
        case -1:
          ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "reactor"), 0);
          /* NOTREACHED */
        case 0:
          ACE_ERROR_RETURN ((LM_ERROR, "(%t) timeout\n"), 0);
          /* NOTREACHED */
        }

      // ACE_DEBUG ((LM_DEBUG, "(%t) done with handle_events\n"));

    }

  ACE_NOTREACHED(return 0);
}

int
ACE_TMAIN (int, ACE_TCHAR *[])
{
  ACE_Reactor *react1 = ACE_Reactor::instance ();
  ACE_Reactor *react2 = new ACE_Reactor ();
  Test_Task tt1[MAX_TASKS];
  Test_Task tt2[MAX_TASKS];

  for (int i = 0; i < MAX_TASKS; i++)
    {
      tt1[i].open (react1);
      tt2[i].open (react2);
    }

  if (ACE_Thread_Manager::instance ()->spawn
      (ACE_THR_FUNC (worker), (void *) react1, THR_NEW_LWP) == -1)
    ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "spawn"), -1);

  else if (ACE_Thread_Manager::instance ()->spawn
      (ACE_THR_FUNC (worker), (void *) react2, THR_NEW_LWP) == -1)
    ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "spawn"), -1);

  ACE_Thread_Manager::instance ()->wait ();
  ACE_DEBUG ((LM_DEBUG, "(%t) done\n"));

  return 0;
}

#else
int
ACE_TMAIN (int, ACE_TCHAR *[])
{
  ACE_ERROR ((LM_ERROR, "threads not supported on this platform\n"));
  return 0;
}
#endif /* ACE_HAS_THREADS */