Merge pull request #203 from apkar/mem-leak

Resolves #202: Fixes memory leak on `ExtConnection`

Author: apkar

src/BufferedConnection.actor.cpp

@@ -40,6 +40,14 @@ struct BCBlock : FastAllocated<BCBlock> /*See below, NonCopyable */ {
 struct BufferedConnectionData {
 	explicit BufferedConnectionData(Reference<IConnection> connection);
 	~BufferedConnectionData() {
+		// Free left over buffers.
+		for (auto blk : deadlist) {
+			delete blk;
+		}
+		for (auto blk : buffer) {
+			delete blk;
+		}
+
 		conn.cancel();
 		connection->close();
 	}
@@ -139,25 +147,6 @@ BufferedConnection::~BufferedConnection() {
 	delete self;
 }
 
-StringRef BufferedConnection::peekSome(int count, int offset) {
-	ASSERT(count + offset <= self->total_bytes.get());
-
-	auto it = self->buffer.begin();
-	int block_offset = self->buffer_begin_offset;
-
-	while (offset) {
-		int advance = std::min(offset, BCBlock::DATA_SIZE - block_offset);
-		offset -= advance;
-		block_offset += advance;
-		if (block_offset == BCBlock::DATA_SIZE) {
-			block_offset = 0;
-			++it;
-		}
-	}
-
-	return {(*it)->data + block_offset, std::min(count, BCBlock::DATA_SIZE - block_offset)};
-}
-
 void BufferedConnectionData::copyInto(uint8_t* buf, int count) {
 	uint8_t* ptr = buf;
 	int offset = buffer_begin_offset;
@@ -307,10 +296,6 @@ Future<Void> BufferedConnection::onWritable() {
 	return self->connection->onWritable();
 }
 
-int BufferedConnection::bytesAvailable() {
-	return self->total_bytes.get();
-}
-
 NetworkAddress BufferedConnection::getPeerAddress() {
 	return self->connection->getPeerAddress();
 }

src/BufferedConnection.h

@@ -37,20 +37,8 @@ struct BufferedConnection : ReferenceCounted<BufferedConnection> {
 	Future<Void> onBytesAvailable(int count);
 
 	/**
-	 * Returns the number of bytes in the buffer.
-	 */
-	int bytesAvailable();
-
-	/**
-	 * Returns up to count bytes starting at offset. The returned memory is
-	 * guaranteed to be valid until the next call to pop() or read().
-	 *
-	 * NOTE: count + offset MUST BE no greater than bytesAvailable().
-	 */
-	StringRef peekSome(int count, int offset = 0);
-
-	/**
-	 * Returns exactly count bytes. The returned memory is guaranteed to be
+	 * Returns exactly count bytes. The returned memor
+	 * y is guaranteed to be
 	 * valid until the next call to pop() or read().
 	 *
 	 * NOTE: count MUST BE no greater than bytesAvailable()

src/Cursor.actor.cpp

@@ -23,21 +23,27 @@
 #include "Knobs.h"
 #include "flow/actorcompiler.h" // This must be the last #include.
 
-int32_t Cursor::prune(std::map<int64_t, Reference<Cursor>>& cursors) {
+int32_t Cursor::prune(std::map<int64_t, Reference<Cursor>>& cursors, bool pruneAll) {
 	time_t now = time(nullptr);
 	int32_t pruned = 0;
 	std::vector<Reference<Cursor>> to_be_pruned;
 
-	for (auto it = cursors.begin(); it != cursors.end();) {
-		if (it->second && now >= it->second->expiry) {
-			to_be_pruned.push_back(it->second);
+	try {
+		for (auto it = cursors.begin(); it != cursors.end();) {
+			if (it->second && (pruneAll || now >= it->second->expiry)) {
+				to_be_pruned.push_back(it->second);
+			}
+			++it;
 		}
-		++it;
-	}
 
-	for (const auto& i : to_be_pruned) {
-		(void)pluck(i);
-		pruned++;
+		for (const auto& i : to_be_pruned) {
+			(void)pluck(i);
+			pruned++;
+		}
+	} catch (Error& e) {
+		TraceEvent(SevError, "BD_cursor_prune_failed").error(e);
+		// Ignoring error just to keep the code consistent with previous behaviour.
+		// Cursor design could be made lot better than this.
 	}
 
 	return pruned;

src/Cursor.h

@@ -43,7 +43,7 @@ struct Cursor : ReferenceCounted<Cursor>, NonCopyable {
 
 	void refresh() { expiry = time(nullptr) + DOCLAYER_KNOBS->CURSOR_EXPIRY; }
 
-	static int32_t prune(std::map<int64_t, Reference<Cursor>>& cursors);
+	static int32_t prune(std::map<int64_t, Reference<Cursor>>& cursors, bool pruneAll);
 
 	static void pluck(Reference<Cursor> cursor);
 	static Reference<Cursor> add(std::map<int64_t, Reference<Cursor>>& siblings, Reference<Cursor> cursor);

src/DocLayer.actor.cpp

@@ -143,7 +143,7 @@ Future<Void> processRequest(Reference<ExtConnection> ec,
 	try {
 		Reference<ExtMsg> msg = ExtMsg::create(header, body, finished);
 		if (verboseLogging)
-			TraceEvent("BD_processRequest").detail("Message", msg->toString());
+			TraceEvent("BD_processRequest").detail("Message", msg->toString()).detail("connId", ec->connectionId);
 		if (verboseConsoleOutput)
 			fprintf(stderr, "C -> S: %s\n\n", msg->toString().c_str());
 		return msg->run(ec);
@@ -165,29 +165,44 @@ ACTOR Future<Void> popDisposedMessages(Reference<BufferedConnection> bc,
 	}
 }
 
+ACTOR Future<Void> housekeeping(Reference<ExtConnection> ec) {
+	try {
+		loop {
+			wait(delay(DOCLAYER_KNOBS->CURSOR_EXPIRY));
+			Cursor::prune(ec->cursors, false);
+		}
+	} catch (Error& e) {
+		// This is the only actor responsible for all the cursors created
+		// through this connection. Prune all the cursors before cancelling
+		// this actor.
+		if (e.code() == error_code_actor_cancelled)
+			Cursor::prune(ec->cursors, true);
+		throw;
+	}
+}
+
 ACTOR Future<Void> extServerConnection(Reference<DocumentLayer> docLayer,
                                        Reference<BufferedConnection> bc,
                                        int64_t connectionId) {
 	if (verboseLogging)
-		TraceEvent("BD_serverNewConnection");
+		TraceEvent("BD_serverNewConnection").detail("connId", connectionId);
 
 	state Reference<ExtConnection> ec = Reference<ExtConnection>(new ExtConnection(docLayer, bc, connectionId));
 	state PromiseStream<std::pair<int, Future<Void>>> msg_size_inuse;
 	state Future<Void> onError = ec->bc->onClosed() || popDisposedMessages(bc, msg_size_inuse.getFuture());
+	state Future<Void> connHousekeeping = housekeeping(ec);
 
 	DocumentLayer::metricReporter->captureGauge(DocLayerConstants::MT_GUAGE_ACTIVE_CONNECTIONS,
 	                                            ++docLayer->nrConnections);
 	try {
-		ec->startHousekeeping();
-
 		loop {
 			// Will be broken (or set or whatever) only when the memory we are passing to processRequest is no longer
 			// needed and can be popped
 			state Promise<Void> finished;
 			choose {
 				when(wait(onError)) {
 					if (verboseLogging)
-						TraceEvent("BD_serverClosedConnection");
+						TraceEvent("BD_serverClosedConnection").detail("connId", connectionId);
 					throw connection_failed();
 				}
 				when(wait(ec->bc->onBytesAvailable(sizeof(ExtMsgHeader)))) {

src/ExtMsg.actor.cpp

@@ -591,7 +591,7 @@ void ExtMsgReply::write(Reference<ExtConnection> nmc) {
 	}
 
 	if (verboseLogging)
-		TraceEvent("BD_msgReply").detail("Message", toString());
+		TraceEvent("BD_msgReply").detail("Message", toString()).detail("connId", nmc->connectionId);
 	if (verboseConsoleOutput)
 		fprintf(stderr, "S -> C: %s\n\n", toString().c_str());
 

src/ExtStructs.actor.cpp

@@ -40,21 +40,6 @@ Reference<Plan> ExtConnection::isolatedWrapOperationPlan(Reference<Plan> plan, i
 	return Reference<Plan>(new RetryPlan(plan, timeout, retryLimit, docLayer->database));
 }
 
-ACTOR Future<Void> housekeeping_impl(Reference<ExtConnection> ec) {
-	loop {
-		wait(delay(DOCLAYER_KNOBS->CURSOR_EXPIRY));
-		try {
-			Cursor::prune(ec->cursors);
-		} catch (Error& e) {
-			TraceEvent(SevError, "BD_Cursor_housekeeping").error(e);
-		}
-	}
-}
-
-void ExtConnection::startHousekeeping() {
-	housekeeping = housekeeping_impl(Reference<ExtConnection>::addRef(this));
-}
-
 ACTOR Future<WriteResult> lastErrorOrLastResult(Future<WriteResult> previous,
                                                 Future<WriteResult> next,
                                                 FlowLock* lock,

src/ExtStructs.h

@@ -113,7 +113,6 @@ struct ExtConnection : ReferenceCounted<ExtConnection>, NonCopyable {
 	Reference<Plan> wrapOperationPlan(Reference<Plan> plan, bool isReadOnly, Reference<UnboundCollectionContext> cx);
 	Reference<Plan> isolatedWrapOperationPlan(Reference<Plan> plan);
 	Reference<Plan> isolatedWrapOperationPlan(Reference<Plan> plan, int64_t timeout, int64_t retryLimit);
-	void startHousekeeping();
 	Future<Void> beforeWrite(int desiredPermits = 1);
 	Future<Void> afterWrite(Future<WriteResult> result, int releasePermits = 1);
 
@@ -152,7 +151,6 @@ struct ExtConnection : ReferenceCounted<ExtConnection>, NonCopyable {
 private:
 	Future<Void> currentWriteLocked;
 	Reference<FlowLock> lock;
-	Future<Void> housekeeping;
 	int32_t maxReceivedRequestID;
 	int32_t nextServerGeneratedRequestID;
 };