New capped collection cursors

2 files changed, 167 insertions, 14 deletions

diff --git a/db/cursor.cpp b/db/cursor.cpp
new file mode 100644
index 00000000000..b9523c09268
--- /dev/null
+++ b/db/cursor.cpp
@@ -0,0 +1,130 @@
+/**
+ *    Copyright (C) 2008 10gen Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "pdfile.h"
+
+class Forward : public AdvanceStrategy {
+    virtual DiskLoc next( const DiskLoc &prev ) const {
+        return prev.rec()->getNext( prev );
+    }
+} _forward;
+
+class Reverse : public AdvanceStrategy {
+    virtual DiskLoc next( const DiskLoc &prev ) const {
+        return prev.rec()->getPrev( prev );
+    }
+} _reverse;
+
+AdvanceStrategy *forward() { return &_forward; }
+AdvanceStrategy *reverse() { return &_reverse; }
+
+DiskLoc nextLoop( NamespaceDetails *nsd, const DiskLoc &prev ) {
+    assert( nsd->capLooped() );
+    DiskLoc next = forward()->next( prev );
+    if ( !next.isNull() )
+        return next;
+    return nsd->firstRecord();
+}
+
+DiskLoc prevLoop( NamespaceDetails *nsd, const DiskLoc &curr ) {
+    assert( nsd->capLooped() );
+    DiskLoc prev = reverse()->next( curr );
+    if ( !prev.isNull() )
+        return prev;
+    return nsd->lastRecord();
+}
+
+ForwardCappedCursor::ForwardCappedCursor( NamespaceDetails *_nsd ) :
+    BasicCursor( DiskLoc(), this ),
+    nsd( _nsd ) {
+    if ( !nsd )
+        return;
+    DiskLoc start;
+    if ( !nsd->capLooped() )
+        start = nsd->firstRecord();
+    else {
+        start = nsd->capExtent.ext()->firstRecord;
+        if ( !start.isNull() && start == nsd->capFirstNewRecord ) {
+            start = nsd->capExtent.ext()->lastRecord;
+            start = nextLoop( nsd, start );
+        }
+    }
+    curr = start;
+}
+
+DiskLoc ForwardCappedCursor::next( const DiskLoc &prev ) const {
+    assert( nsd );
+    if ( !nsd->capLooped() )
+        return forward()->next( prev );
+
+    DiskLoc i = prev;
+    // Last record
+    if ( i == nsd->capExtent.ext()->lastRecord )
+        return DiskLoc();
+    i = nextLoop( nsd, i );
+    // If we become capFirstNewRecord from same extent, advance to next extent.
+    if ( i == nsd->capFirstNewRecord &&
+        i != nsd->capExtent.ext()->firstRecord )
+        i = nextLoop( nsd, nsd->capExtent.ext()->lastRecord );
+    // If we have just gotten to beginning of capExtent, skip to capFirstNewRecord
+    if ( i == nsd->capExtent.ext()->firstRecord )
+        i = nsd->capFirstNewRecord;
+    
+    return i;
+}
+
+ReverseCappedCursor::ReverseCappedCursor( NamespaceDetails *_nsd ) :
+    BasicCursor( DiskLoc(), this ),
+    nsd( _nsd ) {
+    if ( !nsd )
+        return;
+    DiskLoc start;
+    if ( !nsd->capLooped() )
+        start = nsd->lastRecord();
+    else
+        start = nsd->capExtent.ext()->lastRecord;
+    curr = start;
+}
+
+DiskLoc ReverseCappedCursor::next( const DiskLoc &prev ) const {
+    assert( nsd );
+    if ( !nsd->capLooped() )
+        return reverse()->next( prev );
+    
+    DiskLoc i = prev;
+    // Last record
+    if ( nsd->capFirstNewRecord == nsd->capExtent.ext()->firstRecord ) {
+        if ( i == nextLoop( nsd, nsd->capExtent.ext()->lastRecord ) ) {
+            return DiskLoc();
+        }
+    } else {
+        if ( i == nsd->capExtent.ext()->firstRecord ) {
+            return DiskLoc();
+        }
+    }
+    // If we are capFirstNewRecord, advance to prev extent, otherwise just get prev.
+    if ( i == nsd->capFirstNewRecord )
+        i = prevLoop( nsd, nsd->capExtent.ext()->firstRecord );
+    else
+        i = prevLoop( nsd, i );
+    // If we just became last in cap extent, advance past capFirstNewRecord
+    // (We know capExtent.ext()->firstRecord != capFirstNewRecord, since would
+    // have returned DiskLoc() earlier otherwise.)
+    if ( i == nsd->capExtent.ext()->lastRecord )
+        i = reverse()->next( nsd->capFirstNewRecord );
+
+    return i;
+}
\ No newline at end of file
diff --git a/db/cursor.h b/db/cursor.h
index 32aa5a3b4ad..18926902631 100644
--- a/db/cursor.h
+++ b/db/cursor.h
@@ -84,10 +84,19 @@ public:
     }
 };
 
+class AdvanceStrategy {
+public:
+    virtual DiskLoc next( const DiskLoc &prev ) const = 0;
+};
+
+AdvanceStrategy *forward();
+AdvanceStrategy *reverse();
+
 /* table-scan style cursor */
 class BasicCursor : public Cursor {
 protected:
     DiskLoc curr, last;
+    AdvanceStrategy *s;
 
 private:
     // for tailing:
@@ -116,16 +125,16 @@ public:
     bool advance() {
         if ( eof() )
             return false;
-        Record *r = _current();
+        _current();
         last = curr;
-        curr = r->getNext(curr);
+        curr = s->next( curr );
         return ok();
     }
 
-    BasicCursor(DiskLoc dl) : curr(dl) {
+    BasicCursor(DiskLoc dl, AdvanceStrategy *_s = forward()) : curr(dl), s( _s ) {
         init();
     }
-    BasicCursor() {
+    BasicCursor(AdvanceStrategy *_s = forward()) : s( _s ) {
         init();
     }
     virtual string toString() {
@@ -158,19 +167,33 @@ public:
 /* used for order { $natural: -1 } */
 class ReverseCursor : public BasicCursor {
 public:
-    bool advance() {
-        if ( eof() )
-            return false;
-        Record *r = _current();
-        last = curr;
-        curr = r->getPrev(curr);
-        return ok();
+    ReverseCursor(DiskLoc dl) : BasicCursor( dl, reverse() ) { }
+    ReverseCursor() : BasicCursor( reverse() ) { }
+    virtual string toString() {
+        return "ReverseCursor";
     }
+};
+
+class NamespaceDetails;
 
-    ReverseCursor(DiskLoc dl) : BasicCursor(dl) { }
-    ReverseCursor() { }
+class ForwardCappedCursor : public BasicCursor, public AdvanceStrategy {
+public:
+    ForwardCappedCursor( NamespaceDetails *nsd = 0 );
     virtual string toString() {
-        return "ReverseCursor";
+        return "ForwardCappedCursor";
     }
+    virtual DiskLoc next( const DiskLoc &prev ) const;
+private:
+    NamespaceDetails *nsd;
 };
 
+class ReverseCappedCursor : public BasicCursor, public AdvanceStrategy {
+public:
+    ReverseCappedCursor( NamespaceDetails *nsd = 0 );
+    virtual string toString() {
+        return "ReverseCappedCursor";
+    }
+    virtual DiskLoc next( const DiskLoc &prev ) const;
+private:
+    NamespaceDetails *nsd;
+};