/* Web Polygraph       http://www.web-polygraph.org/
 * (C) 2003-2006 The Measurement Factory
 * Licensed under the Apache License, Version 2.0 */

#include "base/polygraph.h"

#include "runtime/globals.h"
#include "client/CltXact.h"
#include "client/SingleCxm.h"


SingleCxm::SingleCxm(CltXact &aXact): theXact(aXact), theFillSize(0) {
}

void SingleCxm::reset() {
	CltXactMgr::reset();
	theFillSize = 0;
}

bool SingleCxm::idle() const {
	return false; // kind of a hack to prevent SingleCxm deletion
}

Connection *SingleCxm::conn() {
	return theXact.conn();
}

void SingleCxm::releaseReadControl(CltXact *x) {
	Assert(&theXact == x);
	// do not stop reading; the new manager will change reservation users
	// conn()->theRd.stop(this);
}

void SingleCxm::control(CltXact *x) {
	Assert(&theXact == x);
	theWrSize = 0;
	theFillSize = 0;
	theXact.conn()->theWr.start(this);
}

void SingleCxm::rewind(CltXact *x) {
	conn()->theRd.stop(this);
	control(x);
}

void SingleCxm::noteAbort(CltXact *x) {
	noteDone(x);
}

void SingleCxm::noteDone(CltXact *) {
	conn()->theWr.stop(this);
	conn()->theRd.stop(this);
}

void SingleCxm::noteLastXaction(CltXact *) {
	// nothing to be done
}

void SingleCxm::noteReadReady(int) {
	// will call us back
	theXact.controlledMasterRead();
}

void SingleCxm::noteWriteReady(int) {
	if (theFillSize <= 0) {
		Assert(conn()->theWrBuf.empty());
		bool needMoreFill = false;
		if (!theXact.controlledFill(needMoreFill))
			return;
		Assert(!needMoreFill); // single transactions must fit or fail
		Assert(conn()->theWrBuf.contSize() > 0);
		theFillSize = conn()->theWrBuf.contSize();
	}

	if (!theXact.controlledMasterWrite(theWrSize))
		return;

	bool needMoreWrite = false;
	if (!theXact.controlledPostWrite(theWrSize, needMoreWrite))
		return;

	if (needMoreWrite) {
		if (!conn()->theWr.theReserv)
			conn()->theWr.start(this);
	} else {
		conn()->theWr.stop(this);
		conn()->theRd.start(this);
		TheFileScanner->setPriority(conn()->sock().fd(), fsupBestEffort);
	}	
}