| 968 | // The next page will be waited for if load is true. |
| 969 | // Only mutex holders will wait on the page read. |
| 970 | ACTOR static Future<Optional<T>> waitThenReadNext(Cursor* self, |
| 971 | Optional<T> inclusiveMaximum, |
| 972 | FlowMutex::Lock* lock, |
| 973 | bool load) { |
| 974 | state FlowMutex::Lock localLock; |
| 975 | |
| 976 | // Lock the mutex if it wasn't already locked, so we didn't get a lock pointer |
| 977 | if (lock == nullptr) { |
| 978 | debug_printf("FIFOQueue::Cursor(%s) waitThenReadNext locking mutex\n", self->toString().c_str()); |
| 979 | FlowMutex::Lock newLock = wait(self->mutex.take()); |
| 980 | localLock = newLock; |
| 981 | } |
| 982 | |
| 983 | if (load) { |
| 984 | debug_printf("FIFOQueue::Cursor(%s) waitThenReadNext waiting for page load\n", |
| 985 | self->toString().c_str()); |
| 986 | wait(success(self->nextPageReader)); |
| 987 | } |
| 988 | |
| 989 | state Optional<T> result = wait(self->readNext(inclusiveMaximum, &localLock)); |
| 990 | |
| 991 | // If a lock was not passed in, so this actor locked the mutex above, then unlock it |
| 992 | if (lock == nullptr) { |
| 993 | // Prevent possible stack overflow if too many waiters which require no IO are queued up |
| 994 | // Using static because multiple Cursors can be involved |
| 995 | static int sinceYield = 0; |
| 996 | if (++sinceYield == 1000) { |
| 997 | sinceYield = 0; |
| 998 | wait(delay(0)); |
| 999 | } |
| 1000 | |
| 1001 | debug_printf("FIFOQueue::Cursor(%s) waitThenReadNext unlocking mutex\n", self->toString().c_str()); |
| 1002 | localLock.release(); |
| 1003 | } |
| 1004 | |
| 1005 | return result; |
| 1006 | } |
| 1007 | |
| 1008 | // Read the next item from the cursor, possibly moving to and waiting for a new page if the prior page was |
| 1009 | // exhausted. If the item is <= inclusiveMaximum, then return it after advancing the cursor to the next item. |