multithreading synchronization issue ......

[GirijaRam]

I have an issue with my code ... my below code is createing databases(database here is a kd tree) and indexing one image per database. i have two classes LastDatabaseTndexingPolicy and another forwardingDatabaseaccessor.cpp .

I am calling the GetDatabaseToAccess() function from forwardingDatabaseAccessor.cpp class .GetDatabaseToAccess() function is present in LastDatabaseTndexingPolicy class and it returns the database created as a pointer and using that database pointer we call another function which actually indexes the image to the database .

Now my issue is i am not able to have multiple threads act on the following functions as DatabaseAccessor_ptr db which is in the following file is coupled with two functions and however i put locks in the LastDatabaseTndexingPolicy file as below i end up getting synchronization issue ..........

Hence now i am using a single lock in forwardingDatabaseAccessor.cpp and serializing the code,.can anyone suggest me how can i change my design to parallelize this code .........

In ForwardingDatabaseAccessor.cpp we are calling function from LastDatabaseTndexingPolicy as shown below:-

DatabaseAccessor_ptr db is something which needs to be synchroinized. I tried createing 256 databases with one image each and when i run this code i ended up creating 175 databses and though i was restricting in code with locks that every database has only one image i ended up having two images in single database ..... ideally i had to get only one image per database but i got two images in few of them hence instead of 256 database this code created 175 or so databases.

code:
indexing::IndexReturnValue ForwardDatabaseAccessor::index(cv::Mat image,
const std::string& imageName, features::Camera::Type indexCameraType,
features::Camera::Type recogCameraType) {
DatabaseAccessor_ptr db = this->IndexingPolicy_ptr->GetDBToIndex();

return db->index(image, imageName, indexCameraType, recogCameraType);
}


In file LastDatabaseTndexingPolicy we have this function GetDatabaseToIndex as follows:----


DatabaseAccessor_ptr LastDatabaseIndexingPolicy::GetDBToIndex()
{
int numDBs;
std::string db_Name;
size_t totNoOfIndexedImagesInDB;
DatabaseAccessor_ptr db;
std::vector<std::string>::iterator endIter;
{
READING_LOCK(lock, adbMutex);
numDBs = this->associateDBs->GetNumberOfAssociateDBs();
endIter = this->associateDBs->end();
endIter--;

}
if (numDBs == 0) {

std::string newDBName = this->CreateAssociateDBs(numDBs);
{
WRITING_LOCK(lock, dbMutex);
db = dbGroup.getDatabase(newDBName);
return db;
}

}

else {


WRITING_LOCK_ADV(writeLock, dbMutex);
totNoOfIndexedImagesInDB = dbGroup.getDatabase(*endIter)->size();
if (totNoOfIndexedImagesInDB >= (this->maxNumberOfImagesDBCFG))
{
writeLock.unlock();
std::string newDBName = this->CreateAssociateDBs(numDBs);
{
WRITING_LOCK(lock, dbMutex);
db = dbGroup.getDatabase(newDBName);
return db;
}
}
else
{

db = dbGroup.getDatabase(*endIter);
return db;
writeLock.unlock();

}
return db;
}
}






std::string IndexToLastDB::CreateAssociateDBs(int numDB) {



std:😱stringstream convert;
convert << (numDB + 1);
std::string newDBName = this->dbName + "_server_fwd_" + convert.str();
std::string db_properties = "type=ripe";
LOG(info,dbName) << "Name of the new database: " << newDBName;
{
WRITING_LOCK(lock, dbMutex);
dbGroup.createDatabase(newDBName, db_properties);
}

{
WRITING_LOCK(lock, adbMutex);
this->associateDBs->AddAssociateDB(newDBName);
}
{
WRITING_LOCK(fileLock, fileMutex);
util::serialization::save<indexing::forward::AssociateDBCollection>(
*this->associateDBs, dbFile.string(), serializationFormat);
}

return newDBName;

}

 

[Cogman]

    indexing::IndexReturnValue ForwardDatabaseAccessor::index(cv::Mat image,
            const std::string& imageName, features::Camera::Type indexCameraType,
            features::Camera::Type recogCameraType) {  
        DatabaseAccessor_ptr db = this->IndexingPolicy_ptr->GetDBToIndex();
        
            return db->index(image, imageName, indexCameraType, recogCameraType);
    }
    
        
        In file LastDatabaseTndexingPolicy we have this function GetDatabaseToIndex as follows:----
        
        
        DatabaseAccessor_ptr LastDatabaseIndexingPolicy::GetDBToIndex()
         {
            int numDBs;
            std::string db_Name;
            size_t totNoOfIndexedImagesInDB;
            DatabaseAccessor_ptr db;
            std::vector<std::string>::iterator endIter;
            {
                READING_LOCK(lock, adbMutex);
                numDBs = this->associateDBs->GetNumberOfAssociateDBs();
                endIter = this->associateDBs->end();
                endIter--;
                
            }
            if (numDBs == 0) {
        
                std::string newDBName = this->CreateAssociateDBs(numDBs);
                {
                    WRITING_LOCK(lock, dbMutex);
                    db = dbGroup.getDatabase(newDBName);
                    return db;
                }
        
            }
        
            else {
        
        
                    WRITING_LOCK_ADV(writeLock, dbMutex);
                    totNoOfIndexedImagesInDB = dbGroup.getDatabase(*endIter)->size();
                    if (totNoOfIndexedImagesInDB >= (this->maxNumberOfImagesDBCFG))
                    {
                        writeLock.unlock();
                        std::string newDBName = this->CreateAssociateDBs(numDBs);
                        {
                        WRITING_LOCK(lock, dbMutex);
                        db = dbGroup.getDatabase(newDBName);
                        return db;
                        }
                    }
                    else
                    {
        
                        db = dbGroup.getDatabase(*endIter);
                        return db;
                        writeLock.unlock();
        
                    }
                    return db;
                }
            }
        
        
        
        
        
        
        std::string IndexToLastDB::CreateAssociateDBs(int numDB) {
        
        
        
                std::ostringstream convert;
                convert << (numDB + 1);
                std::string newDBName = this->dbName + "_server_fwd_" + convert.str();
                std::string db_properties = "type=ripe";
                LOG(info,dbName) << "Name of the new database: " << newDBName;
                {
                    WRITING_LOCK(lock, dbMutex);
                    dbGroup.createDatabase(newDBName, db_properties);
                }
        
                {
                    WRITING_LOCK(lock, adbMutex);
                    this->associateDBs->AddAssociateDB(newDBName);
                }
                {
                    WRITING_LOCK(fileLock, fileMutex);
                    util::serialization::save<indexing::forward::AssociateDBCollection>(
                            *this->associateDBs, dbFile.string(), serializationFormat);
                }
        
                return newDBName;
        
            }

ftfy, please add code tags

Now the advice.

What is "WRITING_LOCK" and "WRITING_LOCK_ADV"?

Next question

                    db = dbGroup.getDatabase(*endIter);
                    return db;
                    writeLock.unlock();

Is that really doing what you think it is doing? looks to me that "writeLock.unlock();" will never be called.

Another issue, if something like "dbGroup.getDatabase(newDBName);" throws an exception, the lock will be maintained (I'm guessing) as it has not explicitly been unlocked.

My suggestion, look into lock_gaurd if you are using c++11, scoped_lock if you aren't using c++11 but are using boost, or implementing those things if you are using neither. Having the lock associated with an object which automatically frees the lock when it leaves scope is much safer in c++ then explicit unlocks.

Also, here is your stack overflow post http://stackoverflow.com/questions/...a-database-and-avoiding-synchronization-issue

 

[GirijaRam]

Thank you soo much for your kind reply....

WRITING_LOCK" is implementing boost lock guard with shared mutex and its designed in such a way such that it gets destroyed when its out of scope .

WRITING_LOCK_ADV is boost unique lock with shared mutex which is implemented in our code such that it allows us to unlock before it gets out of scope , but we cannot unlock WRITING_LOCK before it gets out of scope .

code:

db = dbGroup.getDatabase(*endIter);  
                   return db; 
               writeLock.unlock();

Actually yes WriteLock.unlock(),
even if the above line doesnt get executed as we are returning the DB pointer prior to this line , WriteLock gets destroyed as it goes out of scope .

"Another issue, if something like "dbGroup.getDatabase(newDBName);" throws an exception, the lock will be maintained (I'm guessing) as it has not explicitly been unlocked."

Yes thank you sooo much i should handle exception here as its not been unlocked.

But my real issue is :-

 indexing::IndexReturnValue ForwardDatabaseAccessor::index(cv::Mat image,
            const std::string& imageName, features::Camera::Type indexCameraType,
            features::Camera::Type recogCameraType) {  

        DatabaseAccessor_ptr db = this->IndexingPolicy_ptr->GetDBToIndex();
        
            return db->index(image, imageName, indexCameraType, recogCameraType);
    }

there would be race condition however i lock the database creation code in LastDataIndexingPolicy class as "DatabaseAccessor_ptr db " in the above code in ForwardDatabaseAccessor.cpp class where i would be calling GetDBToIndex() is associated with two functions calls . GetDBToIndex() and Index()

like for example thread 1 is returned a database1 pointer and another thread would also be returning datbase pointer then instead of thread one indexing in database1 it indexs in db2 hence there would be two images in database2 though we have restricted in code for having only one image in each databases. Actually i am sorry i am not able to articulate my issue properly .

 

[EagleKeeper]

Unless the pointer is shared between both threads; each thread should have is own copy in it's own heap.

Thread1(char *database_Label)
{
    Database *db_ptr = new Database(database_label);
    while (db_ptr->Index())
   {
   }
}


Thread2(char *database_Label)
{
    Database *db_ptr = new Database(database_label);
    while (db_ptr->Index())
   {
   }
}

db_ptr is local to each thread due to scope.
Unless you create each thread using the same database; you will not have a problem.

// controlling code
{
void *ptr_Thread1 = CreateThread (Thread1,"database_1");
void *ptr_Thread2 = CreateThread (Thread2,"database_2");

...


}