source: src/Fragmentation/Automation/FragmentScheduler.cpp@ 9875cc

/*
 * Project: MoleCuilder
 * Description: creates and alters molecular systems
 * Copyright (C)  2011 University of Bonn. All rights reserved.
 * Please see the LICENSE file or "Copyright notice" in builder.cpp for details.
 */

/*
 * \file FragmentScheduler.cpp
 *
 * This file strongly follows the Serialization example from the boost::asio
 * library (see server.cpp)
 *
 *  Created on: Oct 19, 2011
 *      Author: heber
 */

// include config.h
#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

// boost asio needs specific operator new
#include <boost/asio.hpp>

#include "CodePatterns/MemDebug.hpp"

#include <boost/bind.hpp>
#include <boost/lexical_cast.hpp>
#include <iostream>
#include <vector>
#include "Connection.hpp" // Must come before boost/serialization headers.
#include <boost/serialization/vector.hpp>
#include "CodePatterns/Info.hpp"
#include "CodePatterns/Log.hpp"
#include "FragmentJob.hpp"
#include "JobId.hpp"

#include "FragmentScheduler.hpp"

FragmentJob FragmentScheduler::NoJob(std::string("NoJob"), JobId::NoJob);

FragmentScheduler::FragmentScheduler(boost::asio::io_service& io_service, unsigned short port) :
  acceptor_(io_service,
      boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port)
  ),
  result(JobId::NoJob)
{
  Info info(__FUNCTION__);
  FragmentJob s(std::string("test"), 1);
  JobsQueue.pushJob(s);

  // Start an accept operation for a new connection.
  connection_ptr new_conn(new Connection(acceptor_.get_io_service()));
  acceptor_.async_accept(new_conn->socket(),
    boost::bind(&FragmentScheduler::handle_accept, this,
      boost::asio::placeholders::error, new_conn));
}

/// Handle completion of a accept operation.
void FragmentScheduler::handle_accept(const boost::system::error_code& e, connection_ptr conn)
{
  Info info(__FUNCTION__);
  if (!e)
  {
    // Successfully accepted a new connection.
    // Check whether there are jobs in the queue
    if (JobsQueue.isJobPresent()) {
      // pop a job and send it to the client.
      FragmentJob job(JobsQueue.popJob());
      // The connection::async_write() function will automatically
      // serialize the data structure for us.
      LOG(1, "INFO: Sending job #" << job.getId() << ".");
      conn->async_write(job,
        boost::bind(&FragmentScheduler::handle_SendJob, this,
        boost::asio::placeholders::error, conn));

      // Start an accept operation for a new Connection only when there
      // are still jobs present otherwise we quit.
      connection_ptr new_conn(new Connection(acceptor_.get_io_service()));
      acceptor_.async_accept(new_conn->socket(),
        boost::bind(&FragmentScheduler::handle_accept, this,
        boost::asio::placeholders::error, new_conn));
    } else {
      // send the static NoJob
      conn->async_write(NoJob,
        boost::bind(&FragmentScheduler::handle_SendJob, this,
        boost::asio::placeholders::error, conn));

      // then there must be no read necesary

      ELOG(2, "There is currently no job present in the queue.");
    }
  }
  else
  {
    // An error occurred. Log it and return. Since we are not starting a new
    // accept operation the io_service will run out of work to do and the
    // server will exit.
    ELOG(0, e.message());
  }
}

/// Callback function when job has been sent.
void FragmentScheduler::handle_SendJob(const boost::system::error_code& e, connection_ptr conn)
{
    Info info(__FUNCTION__);
    LOG(1, "INFO: Job sent.");
    // obtain result
    LOG(1, "INFO: Receiving result for a job ...");
    conn->async_read(result,
      boost::bind(&FragmentScheduler::handle_ReceiveResult, this,
      boost::asio::placeholders::error, conn));
}

/// Callback function when result has been received.
void FragmentScheduler::handle_ReceiveResult(const boost::system::error_code& e, connection_ptr conn)
{
    Info info(__FUNCTION__);
    // nothing to do
    LOG(1, "INFO: Received result for job #" << result.getId() << " ...");
    // and push into queue
    ASSERT(result.getId() != JobId::NoJob,
        "FragmentScheduler::handle_write() - result received has NoJob id.");
    ASSERT(result.getId() != JobId::IllegalJob,
        "FragmentScheduler::handle_write() - result received has IllegalJob id.");
    if ((result.getId() != JobId::NoJob) && (result.getId() != JobId::IllegalJob))
      JobsQueue.pushResult(result);
    // erase result
    result = FragmentResult(JobId::NoJob);
}