bool mts_checkpoint_routine(Relay_log_info *rli, ulonglong period,
                            bool force, bool need_data_lock)
{
  ulong cnt;
  bool error= FALSE;
  struct timespec curr_clock;

  DBUG_ENTER("checkpoint_routine");

#ifndef DBUG_OFF
  if (DBUG_EVALUATE_IF("check_slave_debug_group", 1, 0))
  {
    if (!rli->gaq->count_done(rli))
      DBUG_RETURN(FALSE);
  }
#endif

  /*
    rli->checkpoint_group can have two possible values due to
    two possible status of the last (being scheduled) group. 
  */
  DBUG_ASSERT(!rli->gaq->full() ||
              ((rli->checkpoint_seqno == rli->checkpoint_group -1 &&
                rli->mts_group_status == Relay_log_info::MTS_IN_GROUP) ||
               rli->checkpoint_seqno == rli->checkpoint_group));

  /*
    Currently, the checkpoint routine is being called by the SQL Thread.
    For that reason, this function is called call from appropriate points
    in the SQL Thread's execution path and the elapsed time is calculated
    here to check if it is time to execute it.
  */
  set_timespec_nsec(curr_clock, 0);
  ulonglong diff= diff_timespec(curr_clock, rli->last_clock);
  if (!force && diff < period)
  {
    /*
      We do not need to execute the checkpoint now because
      the time elapsed is not enough.
    */
    DBUG_RETURN(FALSE);
  }

 do
  {
    cnt= rli->gaq->move_queue_head(&rli->workers);
#ifndef DBUG_OFF
    if (DBUG_EVALUATE_IF("check_slave_debug_group", 1, 0) &&
        cnt != opt_mts_checkpoint_period)
      sql_print_error("This an error cnt != mts_checkpoint_period");
#endif
  } while (!sql_slave_killed(rli->info_thd, rli) &&
           cnt == 0 && force &&
           !DBUG_EVALUATE_IF("check_slave_debug_group", 1, 0) &&
           (my_sleep(rli->mts_coordinator_basic_nap), 1));
  /*
    This checks how many consecutive jobs where processed.
    If this value is different than zero the checkpoint
    routine can proceed. Otherwise, there is nothing to be
    done.
  */
  if (cnt == 0)
    goto end;

 /*
    The workers have completed  cnt jobs from the gaq. This means that we
    should increment C->jobs_done by cnt.
  */
  if (!is_mts_worker(rli->info_thd) &&
      !is_mts_db_partitioned(rli))
  {
    DBUG_PRINT("info", ("jobs_done this itr=%ld", cnt));
    static_cast<Mts_submode_logical_clock*>
      (rli->current_mts_submode)->jobs_done+= cnt;
  }

  /* TODO: 
     to turn the least occupied selection in terms of jobs pieces
  */
  for (uint i= 0; i < rli->workers.elements; i++)
  {
    Slave_worker *w_i;
    get_dynamic(&rli->workers, (uchar *) &w_i, i);
    set_dynamic(&rli->least_occupied_workers, (uchar*) &w_i->jobs.len, w_i->id);
  };
  sort_dynamic(&rli->least_occupied_workers, (qsort_cmp) ulong_cmp);

  if (need_data_lock)
    mysql_mutex_lock(&rli->data_lock);
  else
    mysql_mutex_assert_owner(&rli->data_lock);

  /*
    "Coordinator::commit_positions" {

    rli->gaq->lwm has been updated in move_queue_head() and 
    to contain all but rli->group_master_log_name which 
    is altered solely by Coordinator at special checkpoints.
  */
  rli->set_group_master_log_pos(rli->gaq->lwm.group_master_log_pos);
  rli->set_group_relay_log_pos(rli->gaq->lwm.group_relay_log_pos);
  DBUG_PRINT("mts", ("New checkpoint %llu %llu %s",
             rli->gaq->lwm.group_master_log_pos,
             rli->gaq->lwm.group_relay_log_pos,
             rli->gaq->lwm.group_relay_log_name));

  if (rli->gaq->lwm.group_relay_log_name[0] != 0)
    rli->set_group_relay_log_name(rli->gaq->lwm.group_relay_log_name);

  /* 
     todo: uncomment notifies when UNTIL will be supported

     rli->notify_group_master_log_name_update();
     rli->notify_group_relay_log_name_update();

     Todo: optimize with if (wait_flag) broadcast
         waiter: set wait_flag; waits....; drops wait_flag;
  */

  error= rli->flush_info(TRUE);

  mysql_cond_broadcast(&rli->data_cond);
  if (need_data_lock)
    mysql_mutex_unlock(&rli->data_lock);

  /*
    We need to ensure that this is never called at this point when
    cnt is zero. This value means that the checkpoint information
    will be completely reset.
  */
  rli->reset_notified_checkpoint(cnt, rli->gaq->lwm.ts, need_data_lock);

  /* end-of "Coordinator::"commit_positions" */

end:
#ifndef DBUG_OFF
  if (DBUG_EVALUATE_IF("check_slave_debug_group", 1, 0))
    DBUG_SUICIDE();
#endif
  set_timespec_nsec(rli->last_clock, 0);

  DBUG_RETURN(error);
}