Adding new unit dbtests binary, with a few btree tests; moved some global state from db.cpp to instance.cpp
This commit is contained in:
Aaron Staple
345
db/instance.cpp
Normal file
345
db/instance.cpp
Normal file
@@ -0,0 +1,345 @@
|
||||
// instance.cpp : Global state variables and functions.
|
||||
//
|
||||
|
||||
/**
|
||||
* 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 "stdafx.h"
|
||||
#include "db.h"
|
||||
#include "../grid/message.h"
|
||||
#include "../util/mmap.h"
|
||||
#include "../util/hashtab.h"
|
||||
#include "../util/goodies.h"
|
||||
#include "pdfile.h"
|
||||
#include "jsobj.h"
|
||||
#include "javajs.h"
|
||||
#include "query.h"
|
||||
#include "introspect.h"
|
||||
#include "repl.h"
|
||||
#include "../util/unittest.h"
|
||||
#include "dbmessage.h"
|
||||
#include "instance.h"
|
||||
|
||||
bool slave = false;
|
||||
bool master = false; // true means keep an op log
|
||||
extern int curOp;
|
||||
|
||||
boost::mutex dbMutex;
|
||||
int dbLocked = 0;
|
||||
|
||||
int port = DBPort;
|
||||
/* 0 = off; 1 = writes, 2 = reads, 3 = both
|
||||
7 = log a few reads, and all writes.
|
||||
*/
|
||||
int opLogging = 0;
|
||||
int getOpLogging() { return opLogging; }
|
||||
OpLog _oplog;
|
||||
//#define oplog (*(_oplog.f))
|
||||
|
||||
bool useCursors = true;
|
||||
|
||||
void closeAllSockets();
|
||||
void flushOpLog() { _oplog.flush(); }
|
||||
|
||||
void killCursors(int n, long long *ids);
|
||||
void receivedKillCursors(Message& m) {
|
||||
int *x = (int *) m.data->_data;
|
||||
x++; // reserved
|
||||
int n = *x++;
|
||||
assert( n >= 1 );
|
||||
if( n > 2000 ) {
|
||||
problem() << "Assertion failure, receivedKillCursors, n=" << n << endl;
|
||||
assert( n < 30000 );
|
||||
}
|
||||
killCursors(n, (long long *) x);
|
||||
}
|
||||
|
||||
void receivedUpdate(Message& m, stringstream& ss) {
|
||||
DbMessage d(m);
|
||||
const char *ns = d.getns();
|
||||
assert(*ns);
|
||||
setClient(ns);
|
||||
//if( client->profile )
|
||||
ss << ns << ' ';
|
||||
int flags = d.pullInt();
|
||||
BSONObj query = d.nextJsObj();
|
||||
|
||||
assert( d.moreJSObjs() );
|
||||
assert( query.objsize() < m.data->dataLen() );
|
||||
BSONObj toupdate = d.nextJsObj();
|
||||
|
||||
assert( toupdate.objsize() < m.data->dataLen() );
|
||||
assert( query.objsize() + toupdate.objsize() < m.data->dataLen() );
|
||||
updateObjects(ns, toupdate, query, flags & 1, ss);
|
||||
}
|
||||
|
||||
void receivedDelete(Message& m) {
|
||||
DbMessage d(m);
|
||||
const char *ns = d.getns();
|
||||
assert(*ns);
|
||||
setClient(ns);
|
||||
int flags = d.pullInt();
|
||||
bool justOne = flags & 1;
|
||||
assert( d.moreJSObjs() );
|
||||
BSONObj pattern = d.nextJsObj();
|
||||
deleteObjects(ns, pattern, justOne);
|
||||
logOp("d", ns, pattern, 0, &justOne);
|
||||
}
|
||||
|
||||
void receivedQuery(DbResponse& dbresponse, /*AbstractMessagingPort& dbMsgPort, */Message& m, stringstream& ss, bool logit) {
|
||||
MSGID responseTo = m.data->id;
|
||||
|
||||
DbMessage d(m);
|
||||
QueryMessage q(d);
|
||||
|
||||
if( opLogging && logit ) {
|
||||
if( strstr(q.ns, ".$cmd") ) {
|
||||
/* $cmd queries are "commands" and usually best treated as write operations */
|
||||
OPWRITE;
|
||||
}
|
||||
else {
|
||||
OPREAD;
|
||||
}
|
||||
}
|
||||
|
||||
setClient(q.ns);
|
||||
QueryResult* msgdata;
|
||||
|
||||
try {
|
||||
msgdata = runQuery(m, q.ns, q.ntoskip, q.ntoreturn, q.query, q.fields, ss, q.queryOptions);
|
||||
}
|
||||
catch( AssertionException& e ) {
|
||||
ss << " exception ";
|
||||
problem() << " Caught Assertion in runQuery ns:" << q.ns << ' ' << e.toString() << '\n';
|
||||
log() << " ntoskip:" << q.ntoskip << " ntoreturn:" << q.ntoreturn << '\n';
|
||||
if( q.query.valid() )
|
||||
log() << " query:" << q.query.toString() << endl;
|
||||
else
|
||||
log() << " query object is not valid!" << endl;
|
||||
|
||||
BSONObjBuilder err;
|
||||
err.append("$err", e.msg.empty() ? "assertion during query" : e.msg);
|
||||
BSONObj errObj = err.done();
|
||||
|
||||
BufBuilder b;
|
||||
b.skip(sizeof(QueryResult));
|
||||
b.append((void*) errObj.objdata(), errObj.objsize());
|
||||
|
||||
// todo: call replyToQuery() from here instead of this. needs a little tweaking
|
||||
// though to do that.
|
||||
msgdata = (QueryResult *) b.buf();
|
||||
b.decouple();
|
||||
QueryResult *qr = msgdata;
|
||||
qr->_data[0] = 0;
|
||||
qr->_data[1] = 0;
|
||||
qr->_data[2] = 0;
|
||||
qr->_data[3] = 0;
|
||||
qr->len = b.len();
|
||||
qr->setOperation(opReply);
|
||||
qr->cursorId = 0;
|
||||
qr->startingFrom = 0;
|
||||
qr->nReturned = 1;
|
||||
|
||||
}
|
||||
Message *resp = new Message();
|
||||
resp->setData(msgdata, true); // transport will free
|
||||
dbresponse.response = resp;
|
||||
dbresponse.responseTo = responseTo;
|
||||
if( client ) {
|
||||
if( client->profile )
|
||||
ss << " bytes:" << resp->data->dataLen();
|
||||
}
|
||||
else {
|
||||
if( strstr(q.ns, "$cmd") == 0 ) // (this condition is normal for $cmd dropDatabase)
|
||||
log() << "ERROR: receiveQuery: client is null; ns=" << q.ns << endl;
|
||||
}
|
||||
// dbMsgPort.reply(m, resp, responseTo);
|
||||
}
|
||||
|
||||
QueryResult* emptyMoreResult(long long);
|
||||
|
||||
void receivedGetMore(DbResponse& dbresponse, /*AbstractMessagingPort& dbMsgPort, */Message& m, stringstream& ss) {
|
||||
DbMessage d(m);
|
||||
const char *ns = d.getns();
|
||||
ss << ns;
|
||||
setClient(ns);
|
||||
int ntoreturn = d.pullInt();
|
||||
long long cursorid = d.pullInt64();
|
||||
ss << " cid:" << cursorid;
|
||||
ss << " ntoreturn:" << ntoreturn;
|
||||
QueryResult* msgdata;
|
||||
try {
|
||||
msgdata = getMore(ns, ntoreturn, cursorid);
|
||||
}
|
||||
catch( AssertionException& e ) {
|
||||
ss << " exception " + e.toString();
|
||||
msgdata = emptyMoreResult(cursorid);
|
||||
}
|
||||
Message *resp = new Message();
|
||||
resp->setData(msgdata, true);
|
||||
ss << " bytes:" << resp->data->dataLen();
|
||||
ss << " nreturned:" << msgdata->nReturned;
|
||||
dbresponse.response = resp;
|
||||
dbresponse.responseTo = m.data->id;
|
||||
//dbMsgPort.reply(m, resp);
|
||||
}
|
||||
|
||||
void receivedInsert(Message& m, stringstream& ss) {
|
||||
// cout << "GOT MSG id:" << m.data->id << endl;
|
||||
DbMessage d(m);
|
||||
while( d.moreJSObjs() ) {
|
||||
BSONObj js = d.nextJsObj();
|
||||
const char *ns = d.getns();
|
||||
assert(*ns);
|
||||
setClient(ns);
|
||||
ss << ns;
|
||||
theDataFileMgr.insert(ns, (void*) js.objdata(), js.objsize());
|
||||
logOp("i", ns, js);
|
||||
}
|
||||
}
|
||||
|
||||
int ctr = 0;
|
||||
extern int callDepth;
|
||||
|
||||
class JniMessagingPort : public AbstractMessagingPort {
|
||||
public:
|
||||
JniMessagingPort(Message& _container) : container(_container) { }
|
||||
void reply(Message& received, Message& response, MSGID) {
|
||||
container = response;
|
||||
}
|
||||
void reply(Message& received, Message& response) {
|
||||
container = response;
|
||||
}
|
||||
Message & container;
|
||||
};
|
||||
|
||||
/* a call from java/js to the database locally.
|
||||
|
||||
m - inbound message
|
||||
out - outbound message, if there is any, will be set here.
|
||||
if there is one, out.data will be non-null on return.
|
||||
The out.data buffer will automatically clean up when out
|
||||
goes out of scope (out.freeIt==true)
|
||||
|
||||
note we should already be in the mutex lock from connThread() at this point.
|
||||
*/
|
||||
void jniCallback(Message& m, Message& out)
|
||||
{
|
||||
Client *clientOld = client;
|
||||
|
||||
JniMessagingPort jmp(out);
|
||||
callDepth++;
|
||||
int curOpOld = curOp;
|
||||
|
||||
try {
|
||||
|
||||
stringstream ss;
|
||||
char buf[64];
|
||||
time_t_to_String(time(0), buf);
|
||||
buf[20] = 0; // don't want the year
|
||||
ss << buf << " dbjs ";
|
||||
|
||||
{
|
||||
Timer t;
|
||||
|
||||
bool log = false;
|
||||
curOp = m.data->operation();
|
||||
|
||||
if( m.data->operation() == dbQuery ) {
|
||||
// on a query, the Message must have m.freeIt true so that the buffer data can be
|
||||
// retained by cursors. As freeIt is false, we make a copy here.
|
||||
assert( m.data->len > 0 && m.data->len < 32000000 );
|
||||
Message copy(malloc(m.data->len), true);
|
||||
memcpy(copy.data, m.data, m.data->len);
|
||||
DbResponse dbr;
|
||||
receivedQuery(dbr, copy, ss, false);
|
||||
jmp.reply(m, *dbr.response, dbr.responseTo);
|
||||
}
|
||||
else if( m.data->operation() == dbInsert ) {
|
||||
ss << "insert ";
|
||||
receivedInsert(m, ss);
|
||||
}
|
||||
else if( m.data->operation() == dbUpdate ) {
|
||||
ss << "update ";
|
||||
receivedUpdate(m, ss);
|
||||
}
|
||||
else if( m.data->operation() == dbDelete ) {
|
||||
ss << "remove ";
|
||||
receivedDelete(m);
|
||||
}
|
||||
else if( m.data->operation() == dbGetMore ) {
|
||||
DEV log = true;
|
||||
ss << "getmore ";
|
||||
DbResponse dbr;
|
||||
receivedGetMore(dbr, m, ss);
|
||||
jmp.reply(m, *dbr.response, dbr.responseTo);
|
||||
}
|
||||
else if( m.data->operation() == dbKillCursors ) {
|
||||
try {
|
||||
log = true;
|
||||
ss << "killcursors ";
|
||||
receivedKillCursors(m);
|
||||
}
|
||||
catch( AssertionException& ) {
|
||||
problem() << "Caught Assertion in kill cursors, continuing" << endl;
|
||||
ss << " exception ";
|
||||
}
|
||||
}
|
||||
else {
|
||||
cout << " jnicall: operation isn't supported: " << m.data->operation() << endl;
|
||||
assert(false);
|
||||
}
|
||||
|
||||
int ms = t.millis();
|
||||
log = log || ctr++ % 128 == 0;
|
||||
if( log || ms > 100 ) {
|
||||
ss << ' ' << t.millis() << "ms";
|
||||
cout << ss.str().c_str() << endl;
|
||||
}
|
||||
if( client && client->profile >= 1 ) {
|
||||
if( client->profile >= 2 || ms >= 100 ) {
|
||||
// profile it
|
||||
profile(ss.str().c_str()+20/*skip ts*/, ms);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
catch( AssertionException& ) {
|
||||
problem() << "Caught AssertionException in jniCall()" << endl;
|
||||
}
|
||||
|
||||
curOp = curOpOld;
|
||||
callDepth--;
|
||||
|
||||
if( client != clientOld ) {
|
||||
client = clientOld;
|
||||
wassert(false);
|
||||
}
|
||||
}
|
||||
|
||||
#undef exit
|
||||
void dbexit(int rc, const char *why) {
|
||||
log() << " dbexit: " << why << "; flushing op log and files" << endl;
|
||||
flushOpLog();
|
||||
|
||||
/* must do this before unmapping mem or you may get a seg fault */
|
||||
closeAllSockets();
|
||||
|
||||
MemoryMappedFile::closeAllFiles();
|
||||
log() << " dbexit: really exiting now" << endl;
|
||||
exit(rc);
|
||||
}
|
||||
Reference in New Issue
Block a user