add key and data size limit check in db->put. closes #50

git-svn-id: file:///svn/tokudb@832 c7de825b-a66e-492c-adef-691d508d4ae1

newbrt/brt.c

@@ -1483,21 +1483,26 @@ int toku_brt_create(BRT *brt_ptr) {
     return 0;
 }
 
-int toku_brt_set_flags(BRT brt, int flags) {
+int toku_brt_set_flags(BRT brt, unsigned int flags) {
     brt->flags = flags;
     return 0;
 }
 
-int toku_brt_get_flags(BRT brt, int *flags) {
+int toku_brt_get_flags(BRT brt, unsigned int *flags) {
     *flags = brt->flags;
     return 0;
 }
 
-int toku_brt_set_nodesize(BRT brt, int nodesize) {
+int toku_brt_set_nodesize(BRT brt, unsigned int nodesize) {
     brt->nodesize = nodesize;
     return 0;
 }
 
+int toku_brt_get_nodesize(BRT brt, unsigned int *nodesize) {
+    *nodesize = brt->nodesize;
+    return 0;
+}
+
 int toku_brt_set_bt_compare(BRT brt, int (*bt_compare)(DB *, const DBT*, const DBT*)) {
     brt->compare_fun = bt_compare;
     return 0;

newbrt/brt.h

@@ -14,9 +14,10 @@
 int toku_open_brt (const char *fname, const char *dbname, int is_create, BRT *, int nodesize, CACHETABLE, TOKUTXN, int(*)(DB*,const DBT*,const DBT*), DB*);
 
 int toku_brt_create(BRT *);
-int toku_brt_set_flags(BRT, int flags);
-int toku_brt_get_flags(BRT, int *flags);
-int toku_brt_set_nodesize(BRT, int nodesize);
+int toku_brt_set_flags(BRT, unsigned int flags);
+int toku_brt_get_flags(BRT, unsigned int *flags);
+int toku_brt_set_nodesize(BRT, unsigned int nodesize);
+int toku_brt_get_nodesize(BRT, unsigned int *nodesize);
 int toku_brt_set_bt_compare(BRT, int (*bt_compare)(DB *, const DBT*, const DBT*));
 int toku_brt_set_dup_compare(BRT, int (*dup_compare)(DB *, const DBT*, const DBT*));
 int brt_set_cachetable(BRT, CACHETABLE);

src/ydb.c

@@ -574,7 +574,7 @@ static int toku_db_del(DB * db, DB_TXN * txn, DBT * key, u_int32_t flags) {
 static int toku_db_get(DB * db, DB_TXN * txn __attribute__ ((unused)), DBT * key, DBT * data, u_int32_t flags) {
     assert(flags == 0);
     int r;
-    int brtflags;
+    unsigned int brtflags;
     toku_brt_get_flags(db->i->brt, &brtflags);
     if (brtflags & TOKU_DB_DUPSORT) {
         DBC *dbc;
@@ -706,7 +706,27 @@ static int toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *db
 }
 
 static int toku_db_put(DB * db, DB_TXN * txn, DBT * key, DBT * data, u_int32_t flags) {
-    int r = toku_brt_insert(db->i->brt, key, data, txn ? txn->i->tokutxn : 0);
+    int r;
+
+    if (flags != 0)
+        return EINVAL;
+
+    unsigned int brtflags;
+    r = toku_brt_get_flags(db->i->brt, &brtflags); assert(r == 0);
+    unsigned int nodesize;
+    r = toku_brt_get_nodesize(db->i->brt, &nodesize); assert(r == 0);
+
+    if (brtflags & TOKU_DB_DUPSORT) {
+        unsigned int limit = nodesize / (2*BRT_FANOUT-1);
+        if (key->size + data->size >= limit)
+            return EINVAL;
+    } else {
+        unsigned int limit = nodesize / (3*BRT_FANOUT-1);
+        if (key->size >= limit || data->size >= limit)
+            return EINVAL;
+    }
+    
+    r = toku_brt_insert(db->i->brt, key, data, txn ? txn->i->tokutxn : 0);
     //printf("%s:%d %d=__toku_db_put(...)\n", __FILE__, __LINE__, r);
     return r;
 }