Merge pull request #182 from apkar/read_timeouts

Fixing transaction_timed_out errors with long running reads.

Author: apkar

src/QLPlan.actor.cpp

@@ -687,38 +687,59 @@ ACTOR static Future<Void> doNonIsolatedRO(PlanCheckpoint* outerCheckpoint,
                                           Reference<MetadataManager> mm) {
 	if (!dtr)
 		dtr = self->newTransaction();
-	state double startt = now();
 	state Reference<PlanCheckpoint> innerCheckpoint(new PlanCheckpoint);
 	state int nTransactions = 1;
 	state int64_t nResults = 0;
 	state FlowLock* outerLock = outerCheckpoint->getDocumentFinishedLock();
+	state Deque<std::pair<Reference<ScanReturnedContext>, Future<Void>>> bufferedDocs;
+
 	try {
 		state uint64_t metadataVersion = wait(cx->bindCollectionContext(dtr)->getMetadataVersion());
 		loop {
 			state FutureStream<Reference<ScanReturnedContext>> docs = subPlan->execute(innerCheckpoint.getPtr(), dtr);
 			state FlowLock* innerLock = innerCheckpoint->getDocumentFinishedLock();
 			state bool first = true;
-			state Future<Void> timeout = delay(3.0);
+			state Future<Void> timeout = delay(3.0, g_network->getCurrentTask() + 1);
+			state bool finished = false;
 
-			loop choose {
-				when(state Reference<ScanReturnedContext> doc = waitNext(docs)) {
-					// throws end_of_stream when totally finished
-					Void _ = wait(outerLock->take());
-					innerLock->release();
-					output.send(doc);
-					++nResults;
-					if (first) {
-						timeout = delay(DOCLAYER_KNOBS->NONISOLATED_INTERNAL_TIMEOUT);
-						first = false;
+			try {
+				loop choose {
+					when(Reference<ScanReturnedContext> doc = waitNext(docs)) {
+						// throws end_of_stream when totally finished
+						bufferedDocs.push_back(std::make_pair(doc, outerLock->take(g_network->getCurrentTask() + 1)));
+					}
+					when(Void _ = wait(bufferedDocs.empty() ? Never() : bufferedDocs.front().second)) {
+						innerLock->release();
+						output.send(bufferedDocs.front().first);
+						bufferedDocs.pop_front();
+						++nResults;
+						if (first) {
+							timeout =
+							    delay(DOCLAYER_KNOBS->NONISOLATED_INTERNAL_TIMEOUT, g_network->getCurrentTask() + 1);
+							first = false;
+						}
 					}
+					when(Void _ = wait(timeout)) { break; }
 				}
-				when(Void _ = wait(timeout)) { break; }
+				ASSERT(!docs.isReady());
+			} catch (Error& e) {
+				if (e.code() != error_code_end_of_stream)
+					throw;
+				finished = true;
 			}
 
-			ASSERT(!docs.isReady());
-
 			innerCheckpoint = innerCheckpoint->stopAndCheckpoint();
 
+			while (!bufferedDocs.empty()) {
+				Void _ = wait(bufferedDocs.front().second);
+				output.send(bufferedDocs.front().first);
+				bufferedDocs.pop_front();
+				++nResults;
+			}
+
+			if (finished)
+				throw end_of_stream();
+
 			dtr = self->newTransaction();
 			state uint64_t newMetadataVersion = wait(cx->bindCollectionContext(dtr)->getMetadataVersion());
 			if (newMetadataVersion != metadataVersion) {
@@ -763,7 +784,7 @@ ACTOR static Future<Void> doNonIsolatedRW(PlanCheckpoint* outerCheckpoint,
 			state FlowLock* innerLock = innerCheckpoint->getDocumentFinishedLock();
 			state bool first = true;
 			state bool finished = false;
-			state Future<Void> timeout = delay(3.0);
+			state Future<Void> timeout = delay(3.0, g_network->getCurrentTask() + 1);
 			state Deque<std::pair<Reference<ScanReturnedContext>, Future<Void>>> committingDocs;
 			state Deque<Reference<ScanReturnedContext>> bufferedDocs;
 
@@ -781,7 +802,8 @@ ACTOR static Future<Void> doNonIsolatedRW(PlanCheckpoint* outerCheckpoint,
 							         waitNext(docs)) { // throws end_of_stream when totally finished
 								committingDocs.push_back(std::make_pair(doc, doc->commitChanges()));
 								if (first) {
-									timeout = delay(DOCLAYER_KNOBS->NONISOLATED_INTERNAL_TIMEOUT);
+									timeout = delay(DOCLAYER_KNOBS->NONISOLATED_INTERNAL_TIMEOUT,
+									                g_network->getCurrentTask() + 1);
 									first = false;
 								}
 							}
@@ -1148,10 +1170,8 @@ ACTOR static Future<Void> findAndModify(PlanCheckpoint* outerCheckpoint,
                                         PromiseStream<Reference<ScanReturnedContext>> output) {
 	if (!dtr)
 		dtr = NonIsolatedPlan::newTransaction(database);
-	state double startt = now();
 	state Reference<PlanCheckpoint> innerCheckpoint(new PlanCheckpoint);
 	state int nTransactions = 1;
-	state int64_t nResults = 0;
 	state FlowLock* outerLock = outerCheckpoint->getDocumentFinishedLock();
 	state Reference<ScanReturnedContext> firstDoc;
 	state bool any = false;