/* 
 * © 2007-2008 MySQL AB, 2008-2009 Sun Microsystems, Inc.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 * Boston, MA 02111-1307, USA.
 */

#include "stdafx.h"

#include "grt_dispatcher.h"
#include <grtpp_util.h>
#include <grtpp_shell.h>

using namespace bec;

static bool debug_dispatcher= false;

#ifdef __GNUC__
#define DPRINT(fmt, ...) if (debug_dispatcher) g_message(fmt,##__VA_ARGS__)
#else
#define DPRINT(...) if (debug_dispatcher) g_message(__VA_ARGS__)
#endif

class NULLTask : public GRTTaskBase
{
public:
  NULLTask(GRTDispatcher *disp) : GRTTaskBase("Terminate Worker Thread", disp) {}
  
  virtual grt::ValueRef execute(grt::GRT *) { return grt::ValueRef(); }
};


//-----------------------------------------------------------------------------

class GRTSimpleTask : public GRTTaskBase {
public:
  GRTSimpleTask(const std::string &name, GRTDispatcher *disp, const sigc::slot1<grt::ValueRef, grt::GRT*> &function)
    : GRTTaskBase(name, disp), _function(function)
  {
  }

  grt::ValueRef execute(grt::GRT *grt)
  {
    try
    {
      return _function(grt);
    }
    catch (const std::exception &e)
    {
      failed(e);
    }
    return grt::ValueRef();
  }

  virtual void started() {}
  virtual void finished(const grt::ValueRef &result) { set_finished(); }

  virtual void failed(const std::exception &exc)
  {
    const grt::grt_runtime_error *rterr= dynamic_cast<const grt::grt_runtime_error*>(&exc);

    if (rterr)
      _exception= new grt::grt_runtime_error(*rterr);
    else
      _exception= new grt::grt_runtime_error(exc.what(), "");
  }

protected:
  GRTDispatcher *_dispatcher;

private:
  sigc::slot1<grt::ValueRef, grt::GRT*> _function;
};


//---------------------------------------------------------------------------

GRTTaskBase::GRTTaskBase(const std::string &name, GRTDispatcher *disp)
  : _dispatcher(disp), _exception(0), _name(name), _refcount(1), _cancelled(false), _finished(false)
{
}


GRTTaskBase::~GRTTaskBase()
{
  delete _exception;
}


void GRTTaskBase::set_finished()
{
  _finished= true;
}


void GRTTaskBase::cancel()
{
  _cancelled= true;
}


void GRTTaskBase::retain()
{
  _refcount++;
}


void GRTTaskBase::release()
{
  --_refcount;
  
  if (_refcount == 0)
    delete this;
}


void GRTTaskBase::started()
{
  _dispatcher->call_from_main_thread<void>(sigc::mem_fun(this, &GRTTaskBase::started_m), false);
}


void GRTTaskBase::started_m()
{
}


void GRTTaskBase::finished(const grt::ValueRef &result)
{
  retain();
  _dispatcher->call_from_main_thread<void>(sigc::bind<grt::ValueRef>(sigc::mem_fun(this, &GRTTaskBase::finished_m), result),
                                           false);
}


void GRTTaskBase::finished_m(const grt::ValueRef &result)
{
  set_finished();
  release();
}


void GRTTaskBase::failed(const std::exception &exc)
{
  const grt::grt_runtime_error *rterr= dynamic_cast<const grt::grt_runtime_error*>(&exc);

  if (rterr)
    _exception= new grt::grt_runtime_error(*rterr);
  else
    _exception= new grt::grt_runtime_error(exc.what(), "");

  retain();

  _dispatcher->call_from_main_thread<void>(sigc::bind<std::exception>(sigc::mem_fun(this, &GRTTaskBase::failed_m), exc),
                                           false);
}


void GRTTaskBase::failed_m(const std::exception &exc)
{
  set_finished();
  release();
}


struct MutexCentry
{
  MutexCentry(GMutex *mtx) : _mtx(mtx) 
  { 
    g_mutex_lock(mtx);
  }

  ~MutexCentry() 
  { 
    g_mutex_unlock(_mtx);
  }
private:
  GMutex *_mtx;

  MutexCentry(MutexCentry&);
  MutexCentry& operator = (MutexCentry&);
};


void GRTTaskBase::process_message(const grt::Message &msg)
{
  retain();

  _dispatcher->call_from_main_thread<void>(sigc::bind<grt::Message>(sigc::mem_fun(this, &GRTTaskBase::process_message_m), msg),
                                           false); // was true
}


void GRTTaskBase::process_message_m(const grt::Message &msg)
{
  // fallback to the original handlerf
  _default_callbacks.message_cb(msg);

  release();
}


int GRTTaskBase::perform_status_query()
{
  // calling cancel() on the task will cause it to flag the GRT to cancel the task
  return is_cancelled() ? 1 : 0;
}


//---------------------------------------------------------------------------


GRTTask::GRTTask(const std::string &name, GRTDispatcher *owner, const sigc::slot1<grt::ValueRef, grt::GRT*> &function)
  : GRTTaskBase(name, owner), _function(function)
{
}


grt::ValueRef GRTTask::execute(grt::GRT *grt)
{
  return _function(grt);
}


void GRTTask::started_m()
{
  _started.emit();
}


void GRTTask::finished_m(const grt::ValueRef &result)
{
  _finished.emit(result);

  GRTTaskBase::finished_m(result);
}


void GRTTask::failed_m(const std::exception &error)
{
  //_failed.emit(error);
  _failed.emit(*_exception);
  
  //GRTTaskBase::failed_m(error);
  GRTTaskBase::failed_m(*_exception);
}


void GRTTask::process_message_m(const grt::Message &msgs)
{
  if (_message.empty())
  {
    GRTTaskBase::process_message_m(msgs);
    return;
  }

  _message.emit(msgs);

  release();
}

//-----------------------------------------------------------------------------

GRTShellTask::GRTShellTask(const std::string &name, GRTDispatcher *owner, const std::string &command)
  : GRTTaskBase(name, owner)
{
  _command= command;
}


grt::ValueRef GRTShellTask::execute(grt::GRT *grt)
{
  _result= grt->get_shell()->execute(_command);

  _prompt= grt->get_shell()->get_prompt();

  return grt::ValueRef();
}


void GRTShellTask::finished_m(const grt::ValueRef &result)
{
  _finished_signal.emit(_result, _prompt);
  
  GRTTaskBase::finished_m(result);
}

void GRTShellTask::process_message_m(const grt::Message &msg)
{
  _message.emit(msg);

  release();
}

//-----------------------------------------------------------------------------

static void sleep_2ms()
{
  g_usleep(2000);
}

GRTDispatcher::GRTDispatcher(grt::GRT *grt, bool threaded)
  : _busy(false), _threading_disabled(!threaded), _grt(grt)
{
  _worker_running= false;
  _worker_shutdown= false;

  if (threaded)
  {
    _task_queue= g_async_queue_new();

    _callback_queue= g_async_queue_new();
  }
  else
  {
    _task_queue= NULL;
    _callback_queue= NULL;
  }
  _thread= 0;
  
  // default implementation just sleeps for 2ms
  _flush_main_thread_and_wait= sleep_2ms;

  if (getenv("WB_DEBUG_DISPATCHER"))
    debug_dispatcher= true;
}


GRTDispatcher::~GRTDispatcher()
{
  shutdown();

  if (_task_queue)
    g_async_queue_unref(_task_queue);
  if (_callback_queue)
    g_async_queue_unref(_callback_queue);
}


void GRTDispatcher::start()
{
  if (!_threading_disabled)
  {
    DPRINT("starting worker thread");
    _thread= g_thread_create(worker_thread, this, FALSE, NULL);
  }
}


void GRTDispatcher::shutdown()
{
  _worker_shutdown= true;
  if (!_threading_disabled && _worker_running)
  {
    NULLTask *task= new NULLTask(this);
    add_task(task);
    DPRINT("Main thread waiting for worker to finish");
    while (_worker_running) g_usleep(100000);
  }
}


gpointer GRTDispatcher::worker_thread(gpointer data)
{
  GRTDispatcher *self= static_cast<GRTDispatcher*>(data);
  GAsyncQueue *task_queue= self->_task_queue;
  GAsyncQueue *callback_queue= self->_callback_queue;

  DPRINT("worker thread running");
  
  g_async_queue_ref(task_queue);
  g_async_queue_ref(callback_queue);

  self->worker_thread_init();
  
  self->_worker_running= true;
  while (!self->_worker_shutdown)
  {
    GRTTaskBase *task;
    GTimeVal timeout;

    self->_busy= false;
    self->worker_thread_iteration();

    DPRINT("worker: waiting task...");
    // pop next task pushed to queue by the main thread
    
    g_get_current_time(&timeout);
    timeout.tv_sec+= 1;
    task= static_cast<GRTTaskBase*>(g_async_queue_timed_pop(task_queue, &timeout));
    if (!task)
      continue;

    self->_busy= true;
    DPRINT("%s", std::string("worker: got task '"+task->name()+"' handling...").c_str());

    if (dynamic_cast<NULLTask*>(task) != 0) // a NULL task terminates the thread
    {
      DPRINT("worker: termination task received, closing...");
      task->finished(grt::ValueRef());
      task->release();
      break;
    }

    if (task->is_cancelled())
    {
      DPRINT("%s", std::string("worker: task '"+task->name()+"' was cancelled.").c_str());
      task->release();
      continue;
    }

    // do pre-execution preparations
    self->prepare_task(task);

    // execute the task
    self->execute_task(task);

    // cleanup
    task->release();

    DPRINT("worker: task finished.");
  }

  self->worker_thread_release();

  g_async_queue_unref(task_queue);
  g_async_queue_unref(callback_queue);

  self->_worker_running= false;

  DPRINT("worker thread exiting...");

  return NULL;
}


void GRTDispatcher::execute_now(GRTTaskBase *task)
{
  _busy= true;
  prepare_task(task);
  
  execute_task(task);
  
  task->release();
  _busy= false;
}


void GRTDispatcher::add_task(GRTTaskBase *task)
{
  // if threading is disabled or the worker thread is calling another 
  // task, then we have to execute it immediately otherwise we'd just deadlock
  if (_threading_disabled || _thread == g_thread_self())
  {
    execute_now(task);
  }
  else
  {
    g_async_queue_push(_task_queue, task);
  }
}


void GRTDispatcher::cancel_task(GRTTaskBase *task)
{
  task->cancel();
}


bool GRTDispatcher::get_busy()
{
  return (_task_queue && g_async_queue_length(_task_queue) > 0) || _busy;
}


/** Optional callback to wait and flush stuff in main thread.
 *
 * This callback is called when the main thread is waiting on some task to finish.
 * It should at least put the main thread to sleep so that the wait loop
 * doesn't use 100% cpu. It can also perform certain tasks like responding to 
 * screen redraw requests, flushing performSelectorOnMainThread queues (in MacOSX)
 * etc. It should not handle mouse and keyboard events or anything that could
 * cause another call to the backend.
 */
void GRTDispatcher::set_main_thread_flush_and_wait(void (*callback)())
{
  _flush_main_thread_and_wait= callback;
}


void GRTDispatcher::flush_pending_callbacks()
{
  DispatcherCallbackBase *callback;
  
  if (_callback_queue)
  {
    while ((callback= static_cast<DispatcherCallbackBase*>(g_async_queue_try_pop(_callback_queue))))
    {
      callback->execute();
      callback->signal();
      callback->release();  // this release <--------------------------------------------------+
    }                                                                                     //   |
  }                                                                                       //   |
}                                                                                         //   |
                                                                                          //   |
                                                                                          //   |
void GRTDispatcher::call_from_main_thread(DispatcherCallbackBase *callback, bool wait)   //   |
{                                                                                         //   |
  // retain once for ourselves                                                            //   |
  callback->retain();                                                                     //   |
                                                                                          //   |
  // and another time for the main-thread (will be released from main thread)             //   |
  callback->retain();    // this retain matched by... -----------------------------------------+

  if (_threading_disabled || g_thread_self() != _thread)
  {
    callback->execute();
    callback->signal();
    callback->release();
    wait= false;
  }
  else
    g_async_queue_push(_callback_queue, callback);

  if (wait)
    callback->wait();

  callback->release();
}




void GRTDispatcher::worker_thread_init()
{
//QQQ  _grt->enable_thread_notifications();
}

void GRTDispatcher::worker_thread_release()
{
//QQQ  _grt->disable_thread_notifications();
}

void GRTDispatcher::worker_thread_iteration()
{
//QQQ  _grt->flush_notifications();
}


void GRTDispatcher::message_callback(const grt::Message &msgs, GRTTaskBase *task)
{
  task->process_message(msgs);
}


bool GRTDispatcher::status_query_callback(GRTTaskBase *task)
{
  return task->perform_status_query() != 0;
}



void GRTDispatcher::prepare_task(GRTTaskBase *gtask)
{
  gtask->retain();

//  _grt->flush_messages();

  DispatcherCallbacks cb;

  cb.message_cb= 
    _grt->set_message_handler(sigc::bind<GRTTaskBase*>(sigc::mem_fun(this, &GRTDispatcher::message_callback),
                                                       gtask));

  cb.status_query_cb= 
    _grt->set_status_query_handler(sigc::bind<GRTTaskBase*>(sigc::mem_fun(this, &GRTDispatcher::status_query_callback),
                                                            gtask));

  gtask->_default_callbacks= cb;
}


void GRTDispatcher::restore_callbacks(GRTTaskBase *task)
{
//  _grt->flush_messages();

  DispatcherCallbacks &cb(task->_default_callbacks);
  
  _grt->set_message_handler(cb.message_cb);

  _grt->set_status_query_handler(cb.status_query_cb);

  task->release();
}


void GRTDispatcher::execute_task(GRTTaskBase *gtask)
{
  try {
    gtask->started();

    grt::ValueRef result= gtask->execute(_grt);

    gtask->__result= result; // for passing result to add_task_and_wait

    restore_callbacks(gtask);

    gtask->finished(result);

  } catch (std::exception &error) {

    restore_callbacks(gtask);

    gtask->failed(error);
  } 
  catch (std::exception *error) 
  {
    restore_callbacks(gtask);
    gtask->failed(*error);
  }
}


void GRTDispatcher::wait_task(GRTTaskBase *task)
{
  // wait for a task to be completed, making sure that
  // the task won't deadlock because of a call to 
  // call_from_main_thread()

  while (!task->is_finished() && !task->is_cancelled())
  {
    flush_pending_callbacks();
    
    if (_flush_main_thread_and_wait)
      _flush_main_thread_and_wait();
  }
}


grt::ValueRef GRTDispatcher::add_task_and_wait(GRTTaskBase *task) THROW (grt::grt_runtime_error)
{
  grt::ValueRef result;
  
#if 0
  if (is_busy())
  {
    g_error("The GRT dispatcher is currently executing another task.\n"
            "Calling add_task_and_wait() while another task is running can lead\n"
            "to deadlocks and is not allowed.");
    return grt::ValueRef();
  }
#endif

  task->retain();
  
  add_task(task);
  
  // wait for the task to finish executing
  wait_task(task);
  
  // if there was an exception during execution in the worker thread,
  // we throw an exception signaling its failure
  if (task->get_error())
  {
    grt::grt_runtime_error error= *task->get_error();
    task->release();

    throw error;
  }

  result= task->__result;

  task->release();

  return result;
}


grt::ValueRef GRTDispatcher::execute_simple_function(const std::string &name, 
                                                         const sigc::slot1<grt::ValueRef, grt::GRT*> &function) THROW (grt::grt_runtime_error)
{
  GRTSimpleTask *task= new GRTSimpleTask(name, this, function);

  task->retain();

  add_task_and_wait(task);
  
  grt::ValueRef tmp= task->__result;
  
  task->release();
  
  return tmp;
}


void GRTDispatcher::execute_async_function(const std::string &name,
                                           const sigc::slot1<grt::ValueRef, grt::GRT*> &function) THROW (grt::grt_runtime_error)
{
  GRTSimpleTask *task= new GRTSimpleTask(name, this, function);

  task->retain();

  add_task(task);
}

