Changeset 335

Timestamp:

03/15/08 03:04:12 (8 months ago)

Author:

jake

Message:

Another big issue related to not closing the readers explicitly. the sorting struct was being cached

Location:

trunk

Files:

• thrudex/src/CLuceneIndex.cpp (modified) (14 diffs)
• thrudex/src/CLuceneIndex.h (modified) (2 diffs)
• thrudex/src/UpdateFilter.cpp (modified) (1 diff)
• tutorial/perl/BookmarkExample.pl (modified) (7 diffs)

trunk/thrudex/src/CLuceneIndex.cpp

@@ -35 +35 @@
 };
 
+//Used to clean up
+struct reader_deleter
+{
+    void operator()(void const *o) const {
+        IndexReader *r = (IndexReader *)o;
+
+        r->close();
+        delete r;
+    }
+};
+
 CLuceneIndex::CLuceneIndex(const string &index_root, const string &index_name, shared_ptr<Analyzer> analyzer)
     : index_root(index_root), index_name(index_name), analyzer(analyzer), filter_space(10000000), last_synched(0), syncing(false)
@@ -75 +86 @@
         //build up bloom filter
         disk_bloom    = shared_ptr<bloom_filter>(new bloom_filter(filter_space,1.0/(1.0 * filter_space), random_seed));
-        disk_reader   = shared_ptr<IndexReader>(IndexReader::open(idx_path.c_str()));
+        disk_reader   = shared_ptr<IndexReader>(IndexReader::open(idx_path.c_str()), reader_deleter() );
         disk_filter   = shared_ptr<UpdateFilter>(new UpdateFilter(disk_reader));
 
@@ -211 +222 @@
     shared_ptr<set<string> >  l_disk_deletes = disk_deletes;
     shared_ptr<UpdateFilter>  l_disk_filter  = disk_filter;
+    shared_ptr<IndexReader>   l_disk_reader  = disk_reader;
 
     wstring wkey = build_wstring(key);
@@ -249 +261 @@
     shared_ptr<set<string> >  l_disk_deletes = disk_deletes;
     shared_ptr<UpdateFilter>  l_disk_filter  = disk_filter;
+    shared_ptr<IndexReader>   l_disk_reader  = disk_reader;
 
     wstring wkey = build_wstring(key);
@@ -255 +268 @@
     //We'll simply track the docs to remove on next merge
     if( l_disk_bloom->contains( key )){
-        LOG4CXX_DEBUG(logger, "Removed "+key);
+        LOG4CXX_DEBUG(logger, "Removed disk"+key);
         l_disk_deletes->insert( key );
 
@@ -267 +280 @@
     if(l_ram_bloom->contains( key )){
 
+        LOG4CXX_DEBUG(logger, "Removed ram"+key);
+
         Term      *t = new Term(DOC_KEY, wkey.c_str() );
 
@@ -276 +291 @@
         delete t;
     }
+
 }
 
@@ -298 +314 @@
     shared_ptr<IndexSearcher>       l_disk_searcher;
     shared_ptr<UpdateFilter>        l_disk_filter;
+    shared_ptr<IndexReader>         l_disk_reader;
     shared_ptr<MultiSearcher>       l_searcher;
     //RWGuard g(mutex);
@@ -315 +332 @@
 
         l_disk_filter = disk_filter;
+        l_disk_reader = disk_reader;
 
     }
@@ -464 +482 @@
         //RWGuard g(mutex);
         Guard g(mutex);
-        if(last_modified <= last_synched)
+        if(last_modified <= last_synched && disk_deletes->empty())
             return;
     }
@@ -510 +528 @@
     LOG4CXX_DEBUG(logger,"Created Handles");
 
-    //Now we start by deleting any updated docs from disk
-
-    shared_ptr<IndexReader> disk_reader(IndexReader::open(idx_path.c_str()));
-
-    set<string>::iterator it;
-    for( it=l_disk_deletes->begin(); it!=l_disk_deletes->end(); ++it){
-
-        wstring wkey = build_wstring(*it);
-        Term      *t = new Term(DOC_KEY, wkey.c_str() );
-
-        disk_reader->deleteDocuments(t);
-
-        LOG4CXX_DEBUG(logger,"Deleted"+(*it));
-
-        delete t;
-    }
-
-    disk_reader->close();
-
-    LOG4CXX_DEBUG(logger,"Deleted old ids");
+    {
+        //Now we start by deleting any updated docs from disk
+        shared_ptr<IndexReader> tmp_disk_reader(IndexReader::open(idx_path.c_str()));
+
+        int i=0;
+        set<string>::iterator it;
+        for( it=l_disk_deletes->begin(); it!=l_disk_deletes->end(); ++it){
+
+            wstring wkey = build_wstring(*it);
+            Term      *t = new Term(DOC_KEY, wkey.c_str() );
+
+            tmp_disk_reader->deleteDocuments(t);
+
+            LOG4CXX_DEBUG(logger,"Deleted"+(*it));
+
+            delete t;
+            i++;
+        }
+
+        tmp_disk_reader->close();
+
+        LOG4CXX_DEBUG(logger,"Deleted old ids");
+    }
+
 
     {
@@ -552 +574 @@
 
     //Search new index (big perf hit so get it over now)
-    shared_ptr<IndexReader>   l_disk_reader(IndexReader::open(idx_path.c_str()));
-    shared_ptr<IndexSearcher> l_disk_searcher(new IndexSearcher(l_disk_reader.get()));
-    shared_ptr<UpdateFilter>  l_disk_filter(new UpdateFilter(l_disk_reader));
+    shared_ptr<IndexReader>   l_disk_reader( IndexReader::open(idx_path.c_str()), reader_deleter() );
+    shared_ptr<IndexSearcher> l_disk_searcher( new IndexSearcher(l_disk_reader.get()) );
 
 
@@ -560 +581 @@
 
     Query *query = QueryParser::parse( q.c_str(),DOC_KEY,analyzer.get());
-    Hits *h = l_disk_searcher->search(query);
+    Hits  *h     = l_disk_searcher->search(query);
     _CLDELETE(h);
     _CLDELETE(query);
@@ -571 +592 @@
         Guard g(mutex);
 
+        //the order of these things really matters
+        disk_searcher = l_disk_searcher;
+        disk_reader   = l_disk_reader;
+        disk_filter.reset( new UpdateFilter(disk_reader) );
+
+
         //Add any new deletes to the filter
         set<string>::iterator it;
         for( it=disk_deletes->begin(); it!=disk_deletes->end(); ++it){
             wstring wkey = build_wstring(*it);
-            l_disk_filter->skip(wkey);
+
+            LOG4CXX_DEBUG(logger,"Skipping sync:"+*it);
+            disk_filter->skip(wkey);
         }
 
-        disk_reader   = l_disk_reader;
-        disk_filter   = l_disk_filter;
-        disk_searcher = l_disk_searcher;
-
 
         last_synched = Util::currentTime();
 
         syncing = false; //this flag alters the search code to include prev searcher
-    }
-
-
+
+    }
 
     LOG4CXX_DEBUG(logger,"Set new search");
-
-}
+}

trunk/thrudex/src/CLuceneIndex.h

@@ -74 +74 @@
 
     boost::shared_ptr<lucene::index::IndexModifier>  modifier;
-    int64_t                                          last_modified;
+    volatile int64_t                                 last_modified;
 
     boost::shared_ptr<lucene::search::MultiSearcher> searcher;
@@ -80 +80 @@
 
     int64_t                                          last_synched;
-    bool                                             syncing;
+    volatile bool                                    syncing;
 
     boost::shared_ptr<lucene::index::IndexReader>    disk_reader;

trunk/thrudex/src/UpdateFilter.cpp

@@ -42 +42 @@
 void UpdateFilter::skip( wstring key )
 {
-    Term              *t = new Term(DOC_KEY, key.c_str() );
-    TermEnum *enumerator = reader->terms(t);
+
+    TermEnum *enumerator;
+    Term              *t;
+    try{
+    t = new Term(DOC_KEY, key.c_str() );
+    enumerator = reader->terms(t);
     if (enumerator->term(false) == NULL){
         _CLDELETE(enumerator);
         delete t;
         return;
+    }
+    }catch(CLuceneError e){
+        cerr<<"!!!Caught Fatal CLucene Exception: "<<e.what()<<endl;
     }
 

trunk/tutorial/perl/BookmarkExample.pl

@@ -99 +99 @@
 
        $transport->open();
+
+        $self->{thrudoc}->admin("create_bucket", THRUDOC_BUCKET());
     }; if($@) {
         die $@->{message} if UNIVERSAL::isa($@,"Thrift::TException");
@@ -242 +244 @@
     #chunks of 100
     my $limit  = 100;
+    my $i = 0;
     my $r;
     my $seed;
@@ -267 +270 @@
         $seed = $r->{seed};
 
+        $i += @{$r->elements};
+
     }while(@{$r->elements} == $limit);
 
@@ -272 +277 @@
     my $t1 = gettimeofday();
 
-    print "\n*Index cleared in: ".($t1-$t0)."*\n";
+    print "\n*Index cleared($i) in: ".($t1-$t0)."*\n";
 
 }
@@ -322 +327 @@
             }else{
 
-                warn($r->ex->what);
+                warn($r->{ex}->{what});
             }
 
@@ -375 +380 @@
 
 my ($t0,$t1);
+
+
 eval{
 
@@ -386 +393 @@
 
 };if($@){
+
+    warn(Dumper($@));
     die $@->{message} if UNIVERSAL::isa($@,"Thrift::TException");
     die $@;
 }
+