mongodb/db/clientcursor.cpp

/**
*    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/>.
*/

/* clientcursor.cpp

   ClientCursor is a wrapper that represents a cursorid from our database
   application's perspective.

   Cursor -- and its derived classes -- are our internal cursors.
*/

#include "stdafx.h"
#include "query.h"
#include "introspect.h"
#include <time.h>
#include "db.h"

namespace mongo {

    /* TODO: FIX cleanup of clientCursors when hit the end. (ntoreturn insufficient) */

    CCById clientCursorsById;

    /* ------------------------------------------- */

    typedef multimap<DiskLoc, ClientCursor*> ByLoc;
    ByLoc byLoc;
    unsigned byLocSize() {
        return byLoc.size();
    }

    void ClientCursor::setLastLoc(DiskLoc L) {
        if ( L == _lastLoc )
            return;

        if ( !_lastLoc.isNull() ) {
            ByLoc::iterator i = kv_find(byLoc, _lastLoc, this);
            if ( i != byLoc.end() )
                byLoc.erase(i);
        }

        if ( !L.isNull() )
            byLoc.insert( make_pair(L, this) );
        _lastLoc = L;
    }

    /* ------------------------------------------- */

    /* must call this when a btree node is updated */
//void removedKey(const DiskLoc& btreeLoc, int keyPos) {
//}

    /* todo: this implementation is incomplete.  we use it as a prefix for dropDatabase, which
             works fine as the prefix will end with '.'.  however, when used with drop and
    		 deleteIndexes, this could take out cursors that belong to something else -- if you
    		 drop "foo", currently, this will kill cursors for "foobar".
    */
    void ClientCursor::invalidate(const char *nsPrefix) {
        vector<ClientCursor*> toDelete;

        int len = strlen(nsPrefix);
        assert( len > 0 && strchr(nsPrefix, '.') );
        for ( ByLoc::iterator i = byLoc.begin(); i != byLoc.end(); ++i ) {
            ClientCursor *cc = i->second;
            if ( strncmp(nsPrefix, cc->ns.c_str(), len) == 0 )
                toDelete.push_back(i->second);
        }

        for ( vector<ClientCursor*>::iterator i = toDelete.begin(); i != toDelete.end(); ++i )
            delete (*i);
    }

    /* must call when a btree bucket going away.
       note this is potentially slow
    */
    void aboutToDeleteBucket(const DiskLoc& b) {
        RARELY if ( byLoc.size() > 70 ) {
            log() << "perf warning: byLoc.size=" << byLoc.size() << " in aboutToDeleteBucket\n";
        }
        for ( ByLoc::iterator i = byLoc.begin(); i != byLoc.end(); i++ )
            i->second->c->aboutToDeleteBucket(b);
    }

    /* must call this on a delete so we clean up the cursors. */
    void aboutToDelete(const DiskLoc& dl) {
        ByLoc::iterator j = byLoc.lower_bound(dl);
        ByLoc::iterator stop = byLoc.upper_bound(dl);
        if ( j == stop )
            return;

        assert( dbMutexInfo.isLocked() );
        vector<ClientCursor*> toAdvance;

        while ( 1 ) {
            toAdvance.push_back(j->second);
            WIN assert( j->first == dl );
            ++j;
            if ( j == stop )
                break;
        }

        wassert( toAdvance.size() < 5000 );

        for ( vector<ClientCursor*>::iterator i = toAdvance.begin();
                i != toAdvance.end(); ++i )
        {
            Cursor *c = (*i)->c.get();
            DiskLoc tmp1 = c->currLoc();
            if ( tmp1 != dl ) {
                /* this might indicate a failure to call ClientCursor::updateLocation() */
                problem() << "warning: cursor loc does not match byLoc position!" << endl;
            }
            c->checkLocation();
            if ( c->tailing() ) {
                DEV out() << "killing cursor as we would have to advance it and it is tailable" << endl;
                delete *i;
                continue;
            }
            c->advance();
            DiskLoc newLoc = c->currLoc();
            if ( newLoc.isNull() ) {
                // advanced to end -- delete cursor
                delete *i;
            }
            else {
                wassert( newLoc != dl );
                (*i)->updateLocation();
            }
        }
    }

    ClientCursor::~ClientCursor() {
        assert( pos != -2 );
        setLastLoc( DiskLoc() ); // removes us from bylocation multimap
        clientCursorsById.erase(cursorid);
        // defensive:
        (CursorId&) cursorid = -1;
        pos = -2;
    }

    /* call when cursor's location changes so that we can update the
       cursorsbylocation map.  if you are locked and internally iterating, only
       need to call when you are ready to "unlock".
    */
    void ClientCursor::updateLocation() {
        assert( cursorid );
        DiskLoc cl = c->currLoc();
        if ( lastLoc() == cl ) {
            //log() << "info: lastloc==curloc " << ns << '\n';
            return;
        }
        setLastLoc(cl);
        c->noteLocation();
    }

    int ctmLast = 0; // so we don't have to do find() which is a little slow very often.
    long long ClientCursor::allocCursorId() {
        long long x;
        int ctm = (int) curTimeMillis();
        while ( 1 ) {
            x = (((long long)rand()) << 32);
            x = x | ctm | 0x80000000; // OR to make sure not zero
            if ( ctm != ctmLast || ClientCursor::find(x, false) == 0 )
                break;
        }
        ctmLast = ctm;
        DEV out() << "  alloccursorid " << x << endl;
        return x;
    }

    class CursInspector : public SingleResultObjCursor {
        Cursor* clone() {
            return new CursInspector();
        }
        void fill() {
            b.append("byLocation_size", unsigned( byLoc.size() ) );
            b.append("clientCursors_size", unsigned( clientCursorsById.size() ) );
            /* todo update for new impl:
            		stringstream ss;
            		ss << '\n';
            		int x = 40;
            		DiskToCC::iterator it = clientCursorsByLocation.begin();
            		while( it != clientCursorsByLocation.end() ) {
            			DiskLoc dl = it->first;
            			ss << dl.toString() << " -> \n";
            			set<ClientCursor*>::iterator j = it->second.begin();
            			while( j != it->second.end() ) {
            				ss << "    cid:" << j->second->cursorid << ' ' << j->second->ns << " pos:" << j->second->pos << " LL:" << j->second->lastLoc.toString();
            				try {
            					setClient(j->second->ns.c_str());
            					Record *r = dl.rec();
            					ss << " lwh:" << hex << r->lengthWithHeaders << " nxt:" << r->nextOfs << " prv:" << r->prevOfs << dec << ' ' << j->second->c->toString();
            					if( r->nextOfs >= 0 && r->nextOfs < 16 )
            						ss << " DELETED??? (!)";
            				}
            				catch(...) {
            					ss << " EXCEPTION";
            				}
            				ss << "\n";
            				j++;
            			}
            			if( --x <= 0 ) {
            				ss << "only first 40 shown\n" << endl;
            				break;
            			}
            			it++;
            		}
            		b.append("dump", ss.str().c_str());
            */
        }
    public:
        CursInspector() {
            reg("intr.cursors");
        }
    } _ciproto;

} // namespace mongo