/* * 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 #endif // boost asio needs specific operator new #include #include "CodePatterns/MemDebug.hpp" #include #include #include #include #include "Connection.hpp" // Must come before boost/serialization headers. #include #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); }