request.cc

nandflash文件系统源代码

	return (*handler)(prop, res);
}

/*!
 * \brief Various parse handlers.  All of them follow the same model:
 * @param[in] prop: A properties table filled with values sent by the client
 * @param[out] r: If parse is successful, returns a dynamically
 * allocated meta request object. It is the callers responsibility to get rid
 * of this pointer.
 * @retval 0 if parse is successful; -1 otherwise.
 *
 * XXX: Need to make MetaRequest a smart pointer
 */

static int
parseHandlerLookup(Properties &prop, MetaRequest **r)
{
	fid_t dir;
	const char *name;
	seq_t seq;

	seq = prop.getValue("Cseq", (seq_t) -1);
	dir = prop.getValue("Parent File-handle", (fid_t) -1);
	if (dir < 0)
		return -1;
	name = prop.getValue("Filename", (const char*) NULL);
	if (name == NULL)
		return -1;
	*r = new MetaLookup(seq, dir, name);
	return 0;
}

static int
parseHandlerLookupPath(Properties &prop, MetaRequest **r)
{
	fid_t root;
	const char *path;
	seq_t seq;

	seq = prop.getValue("Cseq", (seq_t) -1);
	root = prop.getValue("Root File-handle", (fid_t) -1);
	if (root < 0)
		return -1;
	path = prop.getValue("Pathname", (const char *) NULL);
	if (path == NULL)
		return -1;
	*r = new MetaLookupPath(seq, root, path);
	return 0;
}

static int
parseHandlerCreate(Properties &prop, MetaRequest **r)
{
	fid_t dir;
	const char *name;
	seq_t seq;
	int16_t numReplicas;
	bool exclusive;

	seq = prop.getValue("Cseq", (seq_t) -1);
	dir = prop.getValue("Parent File-handle", (fid_t) -1);
	if (dir < 0)
		return -1;
	name = prop.getValue("Filename", (const char *) NULL);
	if (name == NULL)
		return -1;
	// cap replication
	numReplicas = min((int16_t) prop.getValue("Num-replicas", 1), MAX_REPLICAS_PER_FILE);
	if (numReplicas <= 0)
		return -1;
	// by default, create overwrites the file; when it is turned off,
	// it is for supporting O_EXCL
	exclusive = (prop.getValue("Exclusive", 1)) == 1;

	*r = new MetaCreate(seq, dir, name, numReplicas, exclusive);
	return 0;
}

static int
parseHandlerRemove(Properties &prop, MetaRequest **r)
{
	fid_t dir;
	const char *name;
	seq_t seq;

	seq = prop.getValue("Cseq", (seq_t) -1);
	dir = prop.getValue("Parent File-handle", (fid_t) -1);
	if (dir < 0)
		return -1;
	name = prop.getValue("Filename", (const char *) NULL);
	if (name == NULL)
		return -1;
	*r = new MetaRemove(seq, dir, name);
	return 0;
}

static int
parseHandlerMkdir(Properties &prop, MetaRequest **r)
{
	fid_t dir;
	const char *name;
	seq_t seq;

	seq = prop.getValue("Cseq", (seq_t) -1);
	dir = prop.getValue("Parent File-handle", (fid_t) -1);
	if (dir < 0)
		return -1;
	name = prop.getValue("Directory", (const char *) NULL);
	if (name == NULL)
		return -1;
	*r = new MetaMkdir(seq, dir, name);
	return 0;
}

static int
parseHandlerRmdir(Properties &prop, MetaRequest **r)
{
	fid_t dir;
	const char *name;
	seq_t seq;

	seq = prop.getValue("Cseq", (seq_t) -1);
	dir = prop.getValue("Parent File-handle", (fid_t) -1);
	if (dir < 0)
		return -1;
	name = prop.getValue("Directory", (const char *) NULL);
	if (name == NULL)
		return -1;
	*r = new MetaRmdir(seq, dir, name);
	return 0;
}

static int
parseHandlerReaddir(Properties &prop, MetaRequest **r)
{
	fid_t dir;
	seq_t seq;

	seq = prop.getValue("Cseq", (seq_t) -1);
	dir = prop.getValue("Directory File-handle", (fid_t) -1);
	if (dir < 0)
		return -1;
	*r = new MetaReaddir(seq, dir);
	return 0;
}

static int
parseHandlerReaddirPlus(Properties &prop, MetaRequest **r)
{
	fid_t dir;
	seq_t seq;

	seq = prop.getValue("Cseq", (seq_t) -1);
	dir = prop.getValue("Directory File-handle", (fid_t) -1);
	if (dir < 0)
		return -1;
	*r = new MetaReaddirPlus(seq, dir);
	return 0;
}

static int
parseHandlerGetDirSummary(Properties &prop, MetaRequest **r)
{
	fid_t dir;
	seq_t seq;

	seq = prop.getValue("Cseq", (seq_t) -1);
	dir = prop.getValue("Directory File-handle", (fid_t) -1);
	if (dir < 0)
		return -1;
	*r = new MetaGetDirSummary(seq, dir);
	return 0;
}

static int
parseHandlerGetalloc(Properties &prop, MetaRequest **r)
{
	fid_t fid;
	seq_t seq;
	chunkOff_t offset;

	seq = prop.getValue("Cseq", (seq_t) -1);
	fid = prop.getValue("File-handle", (fid_t) -1);
	offset = prop.getValue("Chunk-offset", (chunkOff_t) -1);
	if ((fid < 0) || (offset < 0))
		return -1;
	*r = new MetaGetalloc(seq, fid, offset);
	return 0;
}

static int
parseHandlerGetlayout(Properties &prop, MetaRequest **r)
{
	fid_t fid;
	seq_t seq;

	seq = prop.getValue("Cseq", (seq_t) -1);
	fid = prop.getValue("File-handle", (fid_t) -1);
	if (fid < 0)
		return -1;
	*r = new MetaGetlayout(seq, fid);
	return 0;
}

static int
parseHandlerAllocate(Properties &prop, MetaRequest **r)
{
	fid_t fid;
	seq_t seq;
	chunkOff_t offset;

	seq = prop.getValue("Cseq", (seq_t) -1);
	fid = prop.getValue("File-handle", (fid_t) -1);
	offset = prop.getValue("Chunk-offset", (chunkOff_t) -1);
	if ((fid < 0) || (offset < 0))
		return -1;
	MetaAllocate *m = new MetaAllocate(seq, fid, offset);
	m->clientHost = prop.getValue("Client-host", "");
	*r = m;
	return 0;
}

static int
parseHandlerTruncate(Properties &prop, MetaRequest **r)
{
	fid_t fid;
	seq_t seq;
	chunkOff_t offset;

	seq = prop.getValue("Cseq", (seq_t) -1);
	fid = prop.getValue("File-handle", (fid_t) -1);
	offset = prop.getValue("Offset", (chunkOff_t) -1);
	if ((fid < 0) || (offset < 0))
		return -1;
	*r = new MetaTruncate(seq, fid, offset);
	return 0;
}

static int
parseHandlerRename(Properties &prop, MetaRequest **r)
{
	fid_t fid;
	seq_t seq;
	const char *oldname;
	const char *newpath;
	bool overwrite;

	seq = prop.getValue("Cseq", (seq_t) -1);
	fid = prop.getValue("Parent File-handle", (fid_t) -1);
	oldname = prop.getValue("Old-name", (const char *) NULL);
	newpath = prop.getValue("New-path", (const char *) NULL);
	overwrite = (prop.getValue("Overwrite", 0)) == 1;
	if ((fid < 0) || (oldname == NULL) || (newpath == NULL))
		return -1;

	*r = new MetaRename(seq, fid, oldname, newpath, overwrite);
	return 0;
}

static int
parseHandlerChangeFileReplication(Properties &prop, MetaRequest **r)
{
	fid_t fid;
	seq_t seq;
	int16_t numReplicas;

	seq = prop.getValue("Cseq", (seq_t) -1);
	fid = prop.getValue("File-handle", (fid_t) -1);
	numReplicas = min((int16_t) prop.getValue("Num-replicas", 1), MAX_REPLICAS_PER_FILE);
	if (numReplicas <= 0)
		return -1;
	*r = new MetaChangeFileReplication(seq, fid, numReplicas);
	return 0;
}

/*!
 * \brief Message that initiates the retiring of a chunkserver.
*/
static int
parseHandlerRetireChunkserver(Properties &prop, MetaRequest **r)
{
	ServerLocation location;
	seq_t seq = prop.getValue("Cseq", (seq_t) -1);
	int downtime;

	location.hostname = prop.getValue("Chunk-server-name", "");
	location.port = prop.getValue("Chunk-server-port", -1);
	if (!location.IsValid()) {
		return -1;
	}
	downtime = prop.getValue("Downtime", -1);
	*r = new MetaRetireChunkserver(seq, location, downtime);
	return 0;
}

static int
parseHandlerToggleRebalancing(Properties &prop, MetaRequest **r)
{
	seq_t seq = prop.getValue("Cseq", (seq_t) -1);
	// 1 is enable; 0 is disable
	int value = prop.getValue("Toggle-rebalancing", 0);
	bool v = (value == 1);

	*r = new MetaToggleRebalancing(seq, v);
	KFS_LOG_VA_INFO("Toggle rebalancing: %d", value);
	return 0;
}

/*!
 * \brief Message that initiates the execution of a rebalance plan.
*/
static int
parseHandlerExecuteRebalancePlan(Properties &prop, MetaRequest **r)
{
	seq_t seq = prop.getValue("Cseq", (seq_t) -1);
	string pathname = prop.getValue("Pathname", "");

	*r = new MetaExecuteRebalancePlan(seq, pathname);
	return 0;
}

/*!
 * \brief Validate that the md5 sent by a chunkserver matches one of the
 * acceptable md5's.
 */
static int
isValidMD5Sum(const string &md5sum)
{
	if (!file_exists(gMD5SumFn)) {
		KFS_LOG_VA_INFO("MD5Sum file %s doesn't exist; no admission control", gMD5SumFn.c_str());
		return 1;
	}
	ifstream ifs;
	const int MAXLINE = 512;
	char line[MAXLINE];

	ifs.open(gMD5SumFn.c_str());

	if (ifs.fail()) {
		KFS_LOG_VA_INFO("Unable to open MD5Sum file %s; no admission control", gMD5SumFn.c_str());
		return 1;
	}

	while (!ifs.eof()) {
		ifs.getline(line, MAXLINE);
		string key = line;
		// remove trailing white space
		string::size_type spc = key.find(' ');
		if (spc != string::npos)
			key.erase(spc);
		if (key == md5sum)
			return 1;
	}

	return 0;
}

/*!
 * \brief Parse out the headers from a HELLO message.  The message
 * body contains the id's of the chunks hosted on the server.
 */
static int
parseHandlerHello(Properties &prop, MetaRequest **r)
{
	seq_t seq = prop.getValue("Cseq", (seq_t) -1);
	MetaHello *hello;
	string key;

	hello = new MetaHello(seq);
	hello->location.hostname = prop.getValue("Chunk-server-name", "");
	hello->location.port = prop.getValue("Chunk-server-port", -1);
	if (!hello->location.IsValid()) {
		delete hello;
		return -1;
	}
	key = prop.getValue("Cluster-key", "");
	if (key != gClusterKey) {
		KFS_LOG_VA_INFO("cluster key mismatch: we have %s, chunkserver sent us %s",
				gClusterKey.c_str(), key.c_str());
		hello->status = -EBADCLUSTERKEY;
	}
	key = prop.getValue("MD5Sum", "");
	if (!isValidMD5Sum(key)) {
		KFS_LOG_VA_INFO("MD5sum mismatch from chunkserver %s:%d: it sent us %s",
				hello->location.hostname.c_str(), hello->location.port, key.c_str());
		hello->status = -EBADCLUSTERKEY;
	}
	hello->totalSpace = prop.getValue("Total-space", (long long) 0);
	hello->usedSpace = prop.getValue("Used-space", (long long) 0);
	hello->rackId = prop.getValue("Rack-id", (int) -1);
	// # of chunks hosted on this server
	hello->numChunks = prop.getValue("Num-chunks", 0);
	// The chunk names follow in the body.  This field tracks
	// the length of the message body
	hello->contentLength = prop.getValue("Content-length", 0);

	*r = hello;
	return 0;
}

/*!
 * \brief Parse out the headers from a LEASE_ACQUIRE message.
 */
int
parseHandlerLeaseAcquire(Properties &prop, MetaRequest **r)
{
	seq_t seq = prop.getValue("Cseq", (seq_t) -1);
	chunkId_t chunkId = prop.getValue("Chunk-handle", (chunkId_t) -1);

	*r = new MetaLeaseAcquire(seq, chunkId);
	return 0;
}

/*!
 * \brief Parse out the headers from a LEASE_RENEW message.
 */
int
parseHandlerLeaseRenew(Properties &prop, MetaRequest **r)
{
	seq_t seq = prop.getValue("Cseq", (seq_t) -1);
	chunkId_t chunkId = prop.getValue("Chunk-handle", (chunkId_t) -1);
	int64_t leaseId = prop.getValue("Lease-id", (int64_t) -1);
	string leaseTypeStr = prop.getValue("Lease-type", "READ_LEASE");
	LeaseType leaseType;

	if (leaseTypeStr == "WRITE_LEASE")
		leaseType = WRITE_LEASE;
	else
		leaseType = READ_LEASE;

	*r = new MetaLeaseRenew(seq, leaseType, chunkId, leaseId);
	return 0;
}

/*!
 * \brief Parse out the headers from a LEASE_RELINQUISH message.
 */
int
parseHandlerLeaseRelinquish(Properties &prop, MetaRequest **r)
{
	seq_t seq = prop.getValue("Cseq", (seq_t) -1);
	chunkId_t chunkId = prop.getValue("Chunk-handle", (chunkId_t) -1);
	int64_t leaseId = prop.getValue("Lease-id", (int64_t) -1);
	string leaseTypeStr = prop.getValue("Lease-type", "READ_LEASE");
	LeaseType leaseType;

	if (leaseTypeStr == "WRITE_LEASE")
		leaseType = WRITE_LEASE;
	else
		leaseType = READ_LEASE;

	*r = new MetaLeaseRelinquish(seq, leaseType, chunkId, leaseId);
	return 0;
}

/*!
 * \brief Parse out the headers from a CORRUPT_CHUNK message.
 */
int
parseHandlerChunkCorrupt(Properties &prop, MetaRequest **r)
{
	seq_t seq = prop.getValue("Cseq", (seq_t) -1);
	fid_t fid = prop.getValue("File-handle", (chunkId_t) -1);
	chunkId_t chunkId = prop.getValue("Chunk-handle", (chunkId_t) -1);

	*r = new MetaChunkCorrupt(seq, fid, chunkId);
	return 0;
}

/*!
 * \brief Parse out the headers from a PING message.
 */
int
parseHandlerPing(Properties &prop, MetaRequest **r)
{
	seq_t seq = prop.getValue("Cseq", (seq_t) -1);

	*r = new MetaPing(seq);
	return 0;
}

/*!
 * \brief Parse out the headers for a UPSERVER message.
 */
int
parseHandlerUpServers(Properties &prop, MetaRequest **r)
{
    seq_t seq = prop.getValue("Cseq", (seq_t) -1);

    *r = new MetaUpServers(seq);
    return 0;
}

/*!
 * \brief Parse out the headers from a TOGGLE_WORM message.
 */
int
parseHandlerToggleWORM(Properties &prop, MetaRequest **r)
{
	seq_t seq = prop.getValue("Cseq", (seq_t) -1);
	// 1 is enable; 0 is disable
	int value = prop.getValue("Toggle-WORM", 0);
	bool v = (value == 1);

	*r = new MetaToggleWORM(seq, v);
	KFS_LOG_VA_INFO("Toggle WORM: %d", value);
	return 0;
}

/*!
 * \brief Parse out the headers from a STATS message.
 */
int
parseHandlerStats(Properties &prop, MetaRequest **r)
{
	seq_t seq = prop.getValue("Cseq", (seq_t) -1);

	*r = new MetaStats(seq);
	return 0;
}

/*!
 * \brief Parse out a dump server map request.
 */
int
parseHandlerDumpChunkToServerMap(Properties &prop, MetaRequest **r)
{
	seq_t seq = prop.getValue("Cseq", (seq_t) -1);

	*r = new MetaDumpChunkToServerMap(seq);
	return 0;
}