mariadb-columnstore-engine/writeengine/redistribute/we_redistributecontrol.cpp

/* Copyright (C) 2014 InfiniDB, Inc.
   Copyright (C) 2016 MariaDB Corporaton

   This program is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License
   as published by the Free Software Foundation; version 2 of
   the License.

   This program 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 General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

/*
* $Id: we_redistributecontrol.cpp 4450 2013-01-21 14:13:24Z rdempsey $
*/

#include <iostream>
#include <set>
#include <vector>
#include <cassert>
#include <stdexcept>
#include <sstream>
#include <string>
#include <ctime>
#include <unistd.h>
//#include <sys/stat.h>
using namespace std;

#include "boost/scoped_ptr.hpp"
#include "boost/scoped_array.hpp"
#include "boost/thread.hpp"
#include "boost/thread/mutex.hpp"
#include "boost/filesystem/path.hpp"
#include "boost/filesystem/operations.hpp"
using namespace boost;

#include "installdir.h"

#include "configcpp.h"
using namespace config;

#include "liboamcpp.h"
using namespace oam;

#include "messagequeue.h"
#include "bytestream.h"
using namespace messageqcpp;

#include "logger.h"

#include "calpontsystemcatalog.h"
using namespace execplan;

#include "we_messages.h"
#include "we_redistributedef.h"
#include "we_redistributecontrolthread.h"
#include "we_redistributeworkerthread.h"
#include "we_redistributecontrol.h"

namespace redistribute
{

RedistributeControl* RedistributeControl::fInstance = NULL;
boost::mutex instanceMutex;

const string RedistributeDir("/data1/systemFiles/redistribute");
const string InfoFileName("/redistribute.info");
const string PlanFileName("/redistribute.plan");


RedistributeControl* RedistributeControl::instance()
{
	// The constructor is protected by instanceMutex lock.
	mutex::scoped_lock lock(instanceMutex);
	if (fInstance == NULL)
		fInstance = new RedistributeControl();

	return fInstance;
}


RedistributeControl::RedistributeControl() : fInfoFilePtr(NULL), fPlanFilePtr(NULL)
{
	// default path /usr/local/mariadb/columnstore/data1/systemFiles/redistribute 
	string installDir = startup::StartUp::installDir();
	fRedistributeDir = installDir + RedistributeDir;
	fInfoFilePath = fRedistributeDir + InfoFileName;
	fPlanFilePath = fRedistributeDir + PlanFileName;

	fOam.reset(new oam::Oam);
	fDbrm.reset(new BRM::DBRM);
	fSysLogger.reset(new logging::Logger(32));  //32 - writeengineserver in SubsystemIDs.txt
	logging::MsgMap msgMap;
	msgMap[logging::M0002] = logging::Message(logging::M0002);
	fSysLogger->msgMap(msgMap);

	//struct stat st;
	//if (stat(fRedistributeDir.c_str(), &st) != 0)
	//filesystem::path dirPath(fRedistributeDir);
	if (filesystem::exists(fRedistributeDir))
	{
		// try to open info file for update if dir exists
		RedistributeInfo info;
		fInfoFilePtr = fopen(fInfoFilePath.c_str(), "r+");
		if (fInfoFilePtr != NULL && 1 == fread(&info, sizeof(info), 1, fInfoFilePtr))
		{
			fRedistributeInfo = info;

			// if there was an active session, mark it as failed until support resume.
			if (fRedistributeInfo.state == RED_STATE_ACTIVE)
				updateState(RED_STATE_FAILED);
		}
	}
}


RedistributeControl::~RedistributeControl()
{
	fOam.reset();
	fDbrm.reset(); 
	delete fInstance;
	fInstance = NULL;
}


int RedistributeControl::handleUIMsg(messageqcpp::ByteStream& bs, messageqcpp::IOSocket& so)
{
	mutex::scoped_lock sessionLock(fSessionMutex);

	uint32_t status = RED_STATE_UNDEF;
	const RedistributeMsgHeader* h = (const RedistributeMsgHeader*) bs.buf();
	try
	{
		switch (h->messageId)
		{
			case RED_CNTL_START:
				status = handleStartMsg(bs, so);
				break;

			case RED_CNTL_STOP:
				status = handleStopMsg(bs, so);
				break;

			case RED_CNTL_CLEAR:
				status = handleClearMsg(bs, so);
				break;

			case RED_CNTL_STATUS:
			default:
				status = handleStatusMsg(bs, so);
				break;
		}
	}
	catch (const std::exception& ex)
	{
		if (fUIResponse.empty())
			fUIResponse = ex.what();
	}
	catch (...)
	{
		if (fUIResponse.empty())
			fUIResponse = "Failed to process the redistribute command.";
	}

	// log the response
	logMessage(fUIResponse);

	//	bs restart() in handlers
	bs.restart();
	bs << (ByteStream::byte) WriteEngine::WE_SVR_REDISTRIBUTE;  // dummy, keep for now.
	bs << status;
	bs << fUIResponse;
	so.write(bs);

	return status;
}


int RedistributeControl::handleStartMsg(messageqcpp::ByteStream& bs, messageqcpp::IOSocket& so)
{
	ostringstream oss;
	uint32_t status = getCurrentState();
	if (status != RED_STATE_IDLE)
	{
		if (status == RED_STATE_ACTIVE)
			oss << "Redistribute is already running.  Command is ignored.  You need to stop and clear this active session before starting a new one.";
		else
			oss << "Redistribute is not in IDLE state.  Command is ignored.  Please check the status of last session, then reset the state to IDLE using action CLEAR.";

		fUIResponse = oss.str();

		return status;
	}

	// must be IDLE state
	try
	{
		// skip the header part, may need save it.
		bs.advance(sizeof(RedistributeMsgHeader));

		getStartOptions(bs);
		RedistributeControlThread::setStopAction(false);
		updateState(RED_STATE_ACTIVE);
	}
	catch (const std::exception& ex)
	{
		if (fErrorMsg.empty())
			fErrorMsg = ex.what();

		fRedistributeInfo.state = RED_STATE_FAILED;
	}
	catch (...)
	{
		fRedistributeInfo.state = RED_STATE_FAILED;
	}

	status = fRedistributeInfo.state;
	if (status == RED_STATE_ACTIVE)
	{
		oss << "Redistribute is started.";
		fControlThread.reset(new boost::thread(RedistributeControlThread(RED_CNTL_START)));
		// Let go the new thread unless we want to call interrupt on this thread in future.
		// Not going to join() because the redistribution could take very long.
		fControlThread->detach();
		fControlThread.reset();
	}
	else
	{
		updateState(RED_STATE_FAILED);
		oss << "Starting redistribute failed.";
		if (!fErrorMsg.empty())
			oss << "  " << fErrorMsg;
	}

	fUIResponse = oss.str();

	return status;
}


int RedistributeControl::handleStatusMsg(messageqcpp::ByteStream&, messageqcpp::IOSocket& so)
{
	ostringstream oss;
	uint32_t status = getCurrentState();
	RedistributeInfo info = fRedistributeInfo;
	switch (status)
	{
		case RED_STATE_IDLE:
			oss << "Redistribute is in IDLE state.";
			break;

		case RED_STATE_ACTIVE:
			oss << "Redistribute is in progress: total " << info.planned;
			if (info.planned > 1)
				oss << " logical partitions are planned to move.\n";
			else
				oss << " logical partition is planned to move.\n";
			if (info.planned > 0)
			{
				if (info.endTime > 0)
					oss << "In " << (info.endTime - info.startTime) << " seconds, ";
				oss << info.success << " success, "
					<< info.skipped << " skipped, "
					<< info.failed << " failed, "
					<< ((info.success+info.skipped+info.failed)*100/info.planned) << "%.";
			}
			break;

		case RED_STATE_FINISH:
			oss << "Redistribute is finished.\n"
				<< info.success << " success, "
				<< info.skipped << " skipped, "
				<< info.failed << " failed.\n";
			if (info.endTime > 0)
				oss << "Total time: " << (info.endTime - info.startTime) << " seconds.\n";
			break;

		case RED_STATE_FAILED:
			oss << "Redistribute is failed.\n";
			try
			{
				size_t l = 0;  // message length
				size_t n = fread(&l, sizeof(int), 1, fInfoFilePtr);
				if (n == 1)
				{
					boost::scoped_array<char> buf(new char[l+1]);
					n = fread(buf.get(), 1, l, fInfoFilePtr);
					if (n == l)
					{
						buf[l] = '\0';
						fErrorMsg += buf.get();
						oss << buf.get();
					}
				}
			}
			catch (const std::exception&)
			{
			}
			catch (...)
			{
			}
			break;

		case RED_STATE_STOPPED:
			oss << "Redistribute is stopped by user.\n";
			if (info.planned > 0)
			{
				if (info.endTime > 0)
					oss << "In " << (info.endTime - info.startTime) << " seconds, ";
				oss << info.success << " success, "
					<< info.skipped << " skipped, "
					<< info.failed << " failed, "
					<< ((info.success+info.skipped+info.failed)*100/info.planned) << "%.";
			}
			break;

		default:
			oss << "Failed to retrieve redistribute information, the file "
				<< fInfoFilePath << " may be corrupted.";
			break;
	}

	fUIResponse = oss.str();

	return status;
}


int RedistributeControl::handleStopMsg(messageqcpp::ByteStream&, messageqcpp::IOSocket& so)
{
	ostringstream oss;
	uint32_t status = getCurrentState();
	if (status != RED_STATE_ACTIVE)
	{
		oss << "Redistribute is not running. Command is ignored.";
	}
	else
	{
		RedistributeControlThread::setStopAction(true);
		updateState(RED_STATE_STOPPED);
		status = RED_STATE_STOPPED;
		boost::thread rct((RedistributeControlThread(RED_CNTL_STOP)));
		rct.join();
		oss << "Redistribute is stopped.";
	}

	fUIResponse = oss.str();

	return status;
}


int RedistributeControl::handleClearMsg(messageqcpp::ByteStream&, messageqcpp::IOSocket& so)
{
	ostringstream oss;
	uint32_t status = getCurrentState();
	if (status == RED_STATE_ACTIVE)
	{
		oss << "Redistribute is running. Command is ignored. To CLEAR, you have to wait or stop the running session.";
	}
	else
	{
		updateState(RED_STATE_IDLE);
		status = RED_STATE_IDLE;
		oss << "Cleared.";
	}

	fUIResponse = oss.str();

	return status;
}


uint32_t RedistributeControl::getCurrentState()
{
	uint32_t status = RED_STATE_UNDEF;
	ostringstream oss;
	mutex::scoped_lock lock(fInfoFileMutex);
	if (!fInfoFilePtr)
	{
		status = RED_STATE_IDLE;
	}
	else
	{
		rewind(fInfoFilePtr);
		RedistributeInfo info;
		size_t n = fread(&info, sizeof(info), 1, fInfoFilePtr);
		if (n == 1)
		{
			fRedistributeInfo = info;
			status = info.state;
		}
	}

	return status;
}


bool RedistributeControl::getStartOptions(messageqcpp::ByteStream& bs)
{
	bool ret = true;
	uint32_t n = 0;
	uint32_t d = 0;

	try
	{
		bs >> fOptions;

		bs >> n;
		fSourceList.clear();
		fSourceList.reserve(n);
		for (uint32_t i = 0; i < n; i++)
		{
			bs >> d;
			fSourceList.push_back(d);
		}
		bs >> n;
		fDestinationList.clear();
		fDestinationList.reserve(n);
		for (uint32_t i = 0; i < n; i++)
		{
			bs >> d;
			fDestinationList.push_back(d);
		}

		if (fSourceList.size() == 0 || fDestinationList.size() == 0)
			throw runtime_error("Failed to get dbroot lists.");
	}
	catch (const std::exception& ex)
	{
		ret = false;
		fErrorMsg = ex.what();
	}
	catch (...)
	{
		ret = false;
		fErrorMsg = "Failed to get dbroot lists.";
	}

	return ret;
}


void RedistributeControl::updateState(uint32_t s)
{
	mutex::scoped_lock lock(fInfoFileMutex);

	// allowed state change:
	//   idle    ->  active
	//   active  ->  finish
	//   active  ->  stopped
	//   active  ->  failed
	//   finish  ->  idle
	//   stopped ->  idle
	//   failed  ->  idle


	if (s == RED_STATE_IDLE)
	{
		if (fRedistributeInfo.state == RED_STATE_ACTIVE)
			return;

		// close the files if they are already opened
		if (fInfoFilePtr != NULL)
		{
			fclose(fInfoFilePtr);
			fInfoFilePtr = NULL;
		}

		if (fPlanFilePtr != NULL)
		{
			fclose(fPlanFilePtr);
			fPlanFilePtr = NULL;
		}

		// move old files to archive
		// zip or compress if the .plan file gets large
		time_t t = fRedistributeInfo.startTime;
		if (t == 0)
			t = time(NULL);

		ostringstream oss;
		struct tm m;
		localtime_r(&t, &m);
		oss << setfill('0') << setw(4) << (m.tm_year+1900) << setw(2) << (m.tm_mon+1)
			<< setw(2) << (m.tm_mday) << setw(2) << (m.tm_hour) << setw(2) << (m.tm_min)
			<< setw(2) << (m.tm_sec);
		try
		{
			if (filesystem::exists(fInfoFilePath) && filesystem::exists(fPlanFilePath))
			{
				bool mergeOk = false;
				FILE* infoPtr = fopen(fInfoFilePath.c_str(), "r+b");
				FILE* entryPtr = fopen(fPlanFilePath.c_str(), "rb");
				int rc = 1;
				if (infoPtr != NULL && entryPtr !=NULL)
				{
					rc = fseek(infoPtr, sizeof(RedistributeInfo), SEEK_SET);
					RedistributePlanEntry entry;
					while (rc == 0)
					{
						size_t n = fread(&entry, sizeof(entry), 1, entryPtr);
						if (n != 1)
							break;

						n = fwrite(&entry, sizeof(entry), 1, infoPtr);
						fflush(infoPtr);
						if (n != 1)
							rc = -1;
					}

				}

				if (rc == 0 && feof(entryPtr))
					mergeOk = true;

				if (infoPtr != NULL)
					fclose(infoPtr);

				if (entryPtr != NULL)
					fclose(entryPtr);

				if (mergeOk)
					filesystem::remove(fPlanFilePath);
			}

			if (filesystem::exists(fInfoFilePath))
			{
				string newInfoPath = fRedistributeDir + "/archive" + InfoFileName + "." + oss.str();
				filesystem::rename(fInfoFilePath, newInfoPath);
			}

			if (filesystem::exists(fPlanFilePath))
			{
				string newPlanPath = fRedistributeDir + "/archive" + PlanFileName + "." + oss.str();
				filesystem::rename(fPlanFilePath, newPlanPath);
			}
		}
		catch (const std::exception&)
		{
		}
		catch (...)
		{
		}

		fRedistributeInfo = RedistributeInfo();
		return;
	}

	// safety check
	if (s != RED_STATE_ACTIVE && fRedistributeInfo.state != RED_STATE_ACTIVE)
		return;

	// in IDLE state there is no redistribute.info file
	if (s == RED_STATE_ACTIVE)
	{
//		filesystem::path dirPath(fRedistributeDir);
//		if (filesystem::exists(fRedistributeDir) && !filesystem::is_directory(fRedistributeDir))
//			filesystem::remove(fRedistributeDir);
		if (!filesystem::exists(fRedistributeDir))
		{
			errno = 0;
			filesystem::create_directory(fRedistributeDir);
			if (!filesystem::exists(fRedistributeDir))
			{
				int e = errno;
				ostringstream oss;
				oss << "Failed to create redistribute directory: ";
				oss << strerror(e) << " (" << e << ")";
				throw runtime_error(oss.str());
			}

			errno = 0;
			filesystem::path archivePath(fRedistributeDir + "/archive");
			filesystem::create_directory(archivePath);
			if (!filesystem::exists(archivePath))
			{
				int e = errno;
				ostringstream oss;
				oss << "Failed to create redistribute archive directory: ";
				oss << strerror(e) << " (" << e << ")";
				throw runtime_error(oss.str());
			}
		}

		fRedistributeInfo.startTime = time(NULL); 
	}


	// open the info file to write
	errno = 0;
	if (fInfoFilePtr == NULL)
		fInfoFilePtr = fopen(fInfoFilePath.c_str(), "w+");
	if (fInfoFilePtr == NULL)
	{
		int e = errno;
		ostringstream oss;
		oss << "Failed to open " << fInfoFilePath << ": " << strerror(e) << " (" << e << ")";
		throw runtime_error(oss.str());
	}

	fRedistributeInfo.state = s;
	if (s == RED_STATE_FINISH)
		fRedistributeInfo.endTime = time(NULL);

	rewind(fInfoFilePtr);
	size_t n = fwrite(&fRedistributeInfo, sizeof(fRedistributeInfo), 1, fInfoFilePtr);
	if (n != 1)
	{
		fclose(fInfoFilePtr);
		fInfoFilePtr = NULL;

		int e = errno;
		ostringstream oss;
		oss << "Failed to write into " << fInfoFilePath << ": " << strerror(e) << " (" << e << ")";
		throw runtime_error(oss.str());
	}

	fflush(fInfoFilePtr);
}


void RedistributeControl::setEntryCount(uint32_t entryCount)
{
	mutex::scoped_lock lock(fInfoFileMutex);
	fRedistributeInfo.planned = entryCount;

	rewind(fInfoFilePtr);
	fwrite(&fRedistributeInfo, sizeof(fRedistributeInfo), 1, fInfoFilePtr);
	fflush(fInfoFilePtr);
}


void RedistributeControl::updateProgressInfo(uint32_t s, time_t t)
{
	mutex::scoped_lock lock(fInfoFileMutex);
	fRedistributeInfo.endTime = t;
	switch(s)
	{
		case RED_TRANS_SUCCESS:
			fRedistributeInfo.success++;
			break;

		case RED_TRANS_SKIPPED:
			fRedistributeInfo.skipped++;
			break;

		default:
			fRedistributeInfo.failed++;
			break;
	}

	rewind(fInfoFilePtr);
	fwrite(&fRedistributeInfo, sizeof(fRedistributeInfo), 1, fInfoFilePtr);
	fflush(fInfoFilePtr);
}


int RedistributeControl::handleJobMsg(messageqcpp::ByteStream& bs, messageqcpp::IOSocket& so)
{
//	mutex::scoped_lock jobLock(fJobMutex);

	uint32_t status = RED_TRANS_SUCCESS;
	try
	{
		fWorkThread.reset(new boost::thread(RedistributeWorkerThread(bs, so)));
		fWorkThread->join();
	}
	catch (const std::exception& ex)
	{
		status = RED_TRANS_FAILED;
		logMessage(ex.what());
	}
	catch (...)
	{
		status = RED_TRANS_FAILED;
	}

	return status;
}


void RedistributeControl::logMessage(const string& msg)
{
    logging::Message::Args args;
	args.add(string("RED:"));
    args.add(msg);

    fSysLogger->logMessage(
			logging::LOG_TYPE_INFO, logging::M0002, args, logging::LoggingID(32, 0, 0));
}


} // namespace

// vim:ts=4 sw=4: