pthread_handler_t handle_slave_worker(void *arg)
{
  THD *thd;                     /* needs to be first for thread_stack */
  bool thd_added= false;
  int error= 0;
  Slave_worker *w= (Slave_worker *) arg;
  Relay_log_info* rli= w->c_rli;
  ulong purge_cnt= 0;
  ulonglong purge_size= 0;
  struct slave_job_item _item, *job_item= &_item;
  Global_THD_manager *thd_manager= Global_THD_manager::get_instance();
  #ifdef HAVE_PSI_INTERFACE
  struct PSI_thread *psi;
  #endif

  my_thread_init();
  DBUG_ENTER("handle_slave_worker");

  thd= new THD;
  if (!thd)
  {
    sql_print_error("Failed during slave worker initialization");
    goto err;
  }
  mysql_mutex_lock(&w->info_thd_lock);
  w->info_thd= thd;
  mysql_mutex_unlock(&w->info_thd_lock);
  thd->thread_stack = (char*)&thd;

  #ifdef HAVE_PSI_INTERFACE
  // save the instrumentation for worker thread in w->info_thd->scheduler
  psi= PSI_THREAD_CALL(get_thread)();
  thd_set_psi(w->info_thd, psi);
  #endif

  pthread_detach_this_thread();
  if (init_slave_thread(thd, SLAVE_THD_WORKER))
  {
    // todo make SQL thread killed
    sql_print_error("Failed during slave worker initialization");
    goto err;
  }
  thd->init_for_queries(w);
  thd_manager->add_thd(thd);
  thd_added= true;

  if (w->update_is_transactional())
  {
    rli->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR, ER(ER_SLAVE_FATAL_ERROR),
                "Error checking if the worker repository is transactional.");
    goto err;
  }

  mysql_mutex_lock(&w->jobs_lock);
  w->running_status= Slave_worker::RUNNING;
  mysql_cond_signal(&w->jobs_cond);

  mysql_mutex_unlock(&w->jobs_lock);

  DBUG_ASSERT(thd->is_slave_error == 0);

  while (!error)
  {
      error= slave_worker_exec_job(w, rli);
  }

  /* 
     Cleanup after an error requires clear_error() go first.
     Otherwise assert(!all) in binlog_rollback()
  */
  thd->clear_error();
  w->cleanup_context(thd, error);

  mysql_mutex_lock(&w->jobs_lock);

  while(de_queue(&w->jobs, job_item))
  {
    purge_cnt++;
    purge_size += ((Log_event*) (job_item->data))->data_written;
    DBUG_ASSERT(job_item->data);
    delete static_cast<Log_event*>(job_item->data);
  }

  DBUG_ASSERT(w->jobs.len == 0);

  mysql_mutex_unlock(&w->jobs_lock);

  mysql_mutex_lock(&rli->pending_jobs_lock);
  rli->pending_jobs -= purge_cnt;
  rli->mts_pending_jobs_size -= purge_size;
  DBUG_ASSERT(rli->mts_pending_jobs_size < rli->mts_pending_jobs_size_max);

  mysql_mutex_unlock(&rli->pending_jobs_lock);

  /*
     In MTS case cleanup_after_session() has be called explicitly.
     TODO: to make worker thd be deleted before Slave_worker instance.
  */
  if (thd->rli_slave)
  {
    w->cleanup_after_session();
    thd->rli_slave= NULL;
  }
  mysql_mutex_lock(&w->jobs_lock);

  w->running_status= Slave_worker::NOT_RUNNING;
  sql_print_information("Worker %lu statistics: "
                        "events processed = %lu "
                        "hungry waits = %lu "
                        "priv queue overfills = %llu ",
                        w->id, w->events_done, w->wq_size_waits_cnt,
                        w->jobs.waited_overfill);
  mysql_cond_signal(&w->jobs_cond);  // famous last goodbye

  mysql_mutex_unlock(&w->jobs_lock);

err:

  if (thd)
  {
    /*
       The slave code is very bad. Notice that it is missing
       several clean up calls here. I've just added what was
       necessary to avoid valgrind errors.
 
       /Alfranio
    */
    DBUG_ASSERT(thd->net.buff != 0);
    net_end(&thd->net);

    /*
      to avoid close_temporary_tables() closing temp tables as those
      are Coordinator's burden.
    */
    thd->system_thread= NON_SYSTEM_THREAD;
    thd->release_resources();

    THD_CHECK_SENTRY(thd);
    if (thd_added)
      thd_manager->remove_thd(thd);
    delete thd;
  }

  my_thread_end();
  ERR_remove_state(0);
  pthread_exit(0);
  DBUG_RETURN(0); 
}