<commit_before>#include "SDBPClient.h"
#include "SDBPControllerConnection.h"
#include "SDBPController.h"
#include "SDBPShardConnection.h"
#include "SDBPPooledShardConnection.h"
#include "SDBPClientConsts.h"
#include "System/Common.h"
#include "System/Macros.h"
#include "System/IO/IOProcessor.h"
#include "System/Events/EventLoop.h"
#include "System/Threading/ThreadPool.h"
#include "System/Threading/LockGuard.h"
#include "Framework/Replication/PaxosLease/PaxosLease.h"
#include "Application/Common/ClientRequest.h"
#include "Application/Common/ClientResponse.h"

// TODO: find out the optimal size
#define MAX_IO_CONNECTION               32768
#define DEFAULT_BATCH_LIMIT             (1*MB)

#ifndef CLIENT_MULTITHREADED
#define CLIENT_MULTITHREADED
#endif

#ifdef CLIENT_MULTITHREADED

// globalMutex protects the underlying single threaded IO and Event handling layer
Mutex                                   globalMutex;                    
unsigned                                numClients = 0;
ThreadPool*                             ioThread = NULL;
Signal                                  ioThreadSignal;

#define CLIENT_MUTEX_GUARD_DECLARE()    MutexGuard clientMutexGuard(mutex)
#define CLIENT_MUTEX_GUARD_LOCK()       clientMutexGuard.Lock()
#define CLIENT_MUTEX_GUARD_UNLOCK()     clientMutexGuard.Unlock()

#define CLIENT_MUTEX_LOCK()             Lock()
#define CLIENT_MUTEX_UNLOCK()           Unlock()

#define GLOBAL_MUTEX_GUARD_DECLARE()    MutexGuard globalMutexGuard(globalMutex)
#define GLOBAL_MUTEX_GUARD_LOCK()       globalMutexGuard.Lock()
#define GLOBAL_MUTEX_GUARD_UNLOCK()     globalMutexGuard.Unlock()

#define GLOBAL_MUTEX_LOCK()             globalMutex.Lock()
#define GLOBAL_MUTEX_UNLOCK()           globalMutex.Unlock()

#else // CLIENT_MULTITHREADED

#define CLIENT_MUTEX_GUARD_DECLARE()
#define CLIENT_MUTEX_GUARD_LOCK()
#define CLIENT_MUTEX_GUARD_UNLOCK()

#define CLIENT_MUTEX_LOCK()
#define CLIENT_MUTEX_UNLOCK()

#define GLOBAL_MUTEX_GUARD_DECLARE()
#define GLOBAL_MUTEX_GUARD_LOCK()
#define GLOBAL_MUTEX_GUARD_UNLOCK()

#define GLOBAL_MUTEX_LOCK()
#define GLOBAL_MUTEX_UNLOCK()

#endif // CLIENT_MULTITHREADED

#define VALIDATE_CONTROLLERS()       \
    if (controller == 0)             \
        return SDBP_API_ERROR;

#define VALIDATE_CONFIG_STATE()     \
    if (configState == NULL)        \
        return SDBP_API_ERROR;

using namespace SDBPClient;

static inline uint64_t Hash(uint64_t h)
{
    return h;
}

static inline uint64_t Key(const ShardConnection* conn)
{
    return conn->GetNodeID();
}

static inline int KeyCmp(uint64_t a, uint64_t b)
{
    if (a < b)
        return -1;
    if (a > b)
        return 1;
    return 0;
}

Client::Client()
{
    controller = NULL;
    globalMutex.SetName("ClientGlobalMutex");
    mutexName.Writef("Client_%p", this);
    mutexName.NullTerminate();
    mutex.SetName(mutexName.GetBuffer());
}

Client::~Client()
{
    Shutdown();
}

int Client::Init(int nodec, const char* nodev[])
{
    // sanity check on parameters
    if (nodec <= 0 || nodev == NULL)
        return SDBP_API_ERROR;

    // set up IO thread, IOProcessor and EventLoop
    GLOBAL_MUTEX_GUARD_DECLARE();
    if (!IOProcessor::Init(MAX_IO_CONNECTION))
        return SDBP_API_ERROR;

    SeedRandomWith(ThreadPool::GetThreadID());

    if (numClients == 0)
    {
        Log_Debug("Creating IOThread");
        ioThreadSignal.SetWaiting(true);
        ioThread = ThreadPool::Create(1);
        ioThread->Execute(MFUNC(Client, IOThreadFunc));
        ioThread->Start();
        ioThreadSignal.Wait();
        Log_Debug("IOThread started");
    }
    numClients++;
    Log_Debug("Number of clients: %u", numClients);
    GLOBAL_MUTEX_GUARD_UNLOCK();

    IOProcessor::BlockSignals(IOPROCESSOR_BLOCK_INTERACTIVE);

    // set default timeouts
    masterTimeout.SetCallable(MFUNC(Client, OnMasterTimeout));
    masterTimeout.SetDelay(3 * PAXOSLEASE_MAX_LEASE_TIME);
    globalTimeout.SetCallable(MFUNC(Client, OnGlobalTimeout));
    globalTimeout.SetDelay(SDBP_DEFAULT_TIMEOUT);

    // set defaults
    commandID = 0;
    result = NULL;
    batchMode = SDBP_BATCH_DEFAULT;
    batchLimit = DEFAULT_BATCH_LIMIT;
    proxy.Init();
    consistencyMode = SDBP_CONSISTENCY_STRICT;
    connectivityStatus = SDBP_NOCONNECTION;
    timeoutStatus = SDBP_SUCCESS;
    transactionStatus = SDBP_SUCCESS;
    numControllerRequests = 0;
    numNestedTransactions = 0;
    transactionQuorumID = 0;
    next = prev = this;

    // create the first result object
    result = new Result;

    // wait for starting properly
    controller = Controller::GetController(this, nodec, nodev);
    if (!controller)
    {
        Shutdown();
        return SDBP_API_ERROR;
    }
    Lock();
    controller->AddClient(this);
    Unlock();

    return SDBP_SUCCESS;
}

void Client::Shutdown()
{
    if (controller == NULL)
        return;
    
    if (batchMode != SDBP_BATCH_NOAUTOSUBMIT)
        Submit();

    isShutdown.SetWaiting(true);
    if (EventLoop::IsRunning())
    {
        onClientShutdown.SetCallable(MFUNC(Client, OnClientShutdown));
        EventLoop::Add(&onClientShutdown);
        isShutdown.Wait();
    }
    EventLoop::Remove(&onClientShutdown);

    GLOBAL_MUTEX_GUARD_DECLARE();
    ASSERT(numClients != 0);
    numClients--;
    if (numClients == 0)
    {
        PooledShardConnection::ShutdownPool();
        
        Log_Debug("Stopping IOThread");
        EventLoop::Stop();
        ioThread->WaitStop();
        delete ioThread;
        ioThread = NULL;
        Log_Debug("IOThread destroyed");
    }

    IOProcessor::Shutdown();

    GLOBAL_MUTEX_GUARD_UNLOCK();

    delete result;
}

// this is always called from the IO thread with the client lock locked
void Client::OnClientShutdown()
{
    ASSERT(numNestedTransactions >= 0);
    if (numNestedTransactions > 0)
        Log_Trace("numNestedTransactions = %d", numNestedTransactions);

    numNestedTransactions = 0;
    ReleaseShardConnections();
    
    EventLoop::Remove(&masterTimeout);
    EventLoop::Remove(&globalTimeout);

    if (!controller->IsShuttingDown())
        Controller::CloseController(this, controller);
    controller = NULL;
    
    shardConnections.DeleteTree();

    DeleteQuorumRequests();
    quorumRequests.Clear();
    numControllerRequests = 0;

    isShutdown.Wake();
}

void Client::SetBatchLimit(unsigned batchLimit_)
{
    batchLimit = batchLimit_;
}

void Client::SetConsistencyMode(int consistencyMode_)
{
    consistencyMode = consistencyMode_;
}

void Client::SetBatchMode(int batchMode_)
{
    batchMode = batchMode_;
}

void Client::SetGlobalTimeout(uint64_t timeout)
{
    if (globalTimeout.IsActive())
    {
        EventLoop::Remove(&globalTimeout);
        globalTimeout.SetDelay(timeout);
        EventLoop::Add(&globalTimeout);
    }
    else
        globalTimeout.SetDelay(timeout);
}

void Client::SetMasterTimeout(uint64_t timeout)
{
    Log_Trace("timeout: %U", timeout);

    if (masterTimeout.IsActive())
    {
        EventLoop::Remove(&masterTimeout);
        masterTimeout.SetDelay(timeout);
        EventLoop::Add(&masterTimeout);
    }
    else
        masterTimeout.SetDelay(timeout);
}

uint64_t Client::GetGlobalTimeout()
{
    return globalTimeout.GetDelay();
}

uint64_t Client::GetMasterTimeout()
{
    return masterTimeout.GetDelay();
}

// this must be called with the client lock locked!
ConfigState* Client::UnprotectedGetConfigState()
{
    if (!configState.hasMaster)
    {
        result->Close();
        Unlock();
        EventLoop();
        Lock();
    }

    return &configState;
}

ConfigState* Client::GetConfigState()
{
    Log_Trace();

    CLIENT_MUTEX_GUARD_DECLARE();

    if (controller == NULL)
        return NULL;
    
    if (!configState.hasMaster)
    {
        result->Close();
        CLIENT_MUTEX_GUARD_UNLOCK();
        EventLoop();
        CLIENT_MUTEX_GUARD_LOCK();
    }

    return &configState;
}

void Client::CloneConfigState(ConfigState& configState_)
{
    CLIENT_MUTEX_GUARD_DECLARE();

    if (controller == NULL)
        return;
    
    if (!configState.hasMaster)
    {
        result->Close();
        CLIENT_MUTEX_GUARD_UNLOCK();
        EventLoop();
        CLIENT_MUTEX_GUARD_LOCK();
    }

    configState_ = configState;
}

void Client::WaitConfigState()
{
    CLIENT_MUTEX_GUARD_DECLARE();

    if (controller == NULL)
        return;

    // delete configState, so EventLoop() will wait for a new one
    configState.hasMaster = false;
    result->Close();
    CLIENT_MUTEX_GUARD_UNLOCK();
    EventLoop();
    CLIENT_MUTEX_GUARD_LOCK();
}

Result* Client::GetResult()
{
    CLIENT_MUTEX_GUARD_DECLARE();

    if (controller == NULL)
        return NULL;

    Result* tmp;
    
    tmp = result;
    result = new Result;
    return tmp;
}

int Client::GetTransportStatus()
{
    VALIDATE_CONTROLLERS();
    return result->GetTransportStatus();
}

int Client::GetConnectivityStatus()
{
    return connectivityStatus;
}

int Client::GetTimeoutStatus()
{
    return timeoutStatus;
}

int Client::GetCommandStatus()
{
    VALIDATE_CONTROLLERS();
    return result->GetCommandStatus();
}

int Client::CreateDatabase(ReadBuffer& name)
{
    Request* req;

    req = new Request;
    req->CreateDatabase(controller->NextCommandID(), name);

    return ConfigRequest(req);
}

int Client::RenameDatabase(uint64_t databaseID, const ReadBuffer& name)
{
    Request* req;

    req = new Request;
    req->RenameDatabase(controller->NextCommandID(), databaseID, (ReadBuffer&) name);

    return ConfigRequest(req);
}

int Client::DeleteDatabase(uint64_t databaseID)
{
    Request* req;

    req = new Request;
    req->DeleteDatabase(controller->NextCommandID(), databaseID);

    return ConfigRequest(req);
}

int Client::CreateTable(uint64_t databaseID, uint64_t quorumID, ReadBuffer& name)
{
    Request* req;

    req = new Request;
    req->CreateTable(controller->NextCommandID(), databaseID, quorumID, name);

    return ConfigRequest(req);
}

int Client::RenameTable(uint64_t tableID, ReadBuffer& name)
{
    Request* req;

    req = new Request;
    req->RenameTable(controller->NextCommandID(), tableID, name);

    return ConfigRequest(req);
}

int Client::DeleteTable(uint64_t tableID)
{
    Request* req;

    req = new Request;
    req->DeleteTable(controller->NextCommandID(), tableID);

    return ConfigRequest(req);
}

int Client::TruncateTable(uint64_t tableID)
{
    Request* req;

    req = new Request;
    req->TruncateTable(controller->NextCommandID(), tableID);

    return ConfigRequest(req);
}

#define GET_CONFIG_STATE_OR_RETURN(...) \
    CLIENT_MUTEX_GUARD_DECLARE();       \
                                        \
    if (controller == NULL)             \
        return __VA_ARGS__;             \
                                        \
    if (!configState.hasMaster)         \
    {                                   \
        result->Close();                \
        CLIENT_MUTEX_GUARD_UNLOCK();    \
        EventLoop();                    \
        CLIENT_MUTEX_GUARD_LOCK();      \
    }                                   \
    if (!configState.hasMaster)         \
        return __VA_ARGS__;

unsigned Client::GetNumQuorums()
{
    GET_CONFIG_STATE_OR_RETURN(0);
    
    return configState.quorums.GetLength();
}

uint64_t Client::GetQuorumIDAt(unsigned n)
{
    ConfigQuorum*   quorum;
    unsigned        i;

    GET_CONFIG_STATE_OR_RETURN(0);
    
    i = 0;
    FOREACH (quorum, configState.quorums)
    {
        if (i == n)
            return quorum->quorumID;
            
        i++;
    }

    return 0;
}

void Client::GetQuorumNameAt(unsigned n, Buffer& name)
{
    ConfigQuorum*   quorum;
    unsigned        i;

    GET_CONFIG_STATE_OR_RETURN();
    
    i = 0;
    FOREACH (quorum, configState.quorums)
    {
        if (i == n)
        {
            name.Write(quorum->name);
            return;
        }
            
        i++;
    }
}

unsigned Client::GetNumDatabases()
{
    GET_CONFIG_STATE_OR_RETURN(0);
    
    return configState.databases.GetLength();
}

uint64_t Client::GetDatabaseIDAt(unsigned n)
{
    ConfigDatabase* database;
    unsigned        i;

    GET_CONFIG_STATE_OR_RETURN(0);
    
    i = 0;
    FOREACH (database, configState.databases)
    {
        if (i == n)
            return database->databaseID;
            
        i++;
    }

    return 0;
}

void Client::GetDatabaseNameAt(unsigned n, Buffer& name)
{
    ConfigDatabase* database;
    unsigned        i;

    GET_CONFIG_STATE_OR_RETURN();

    i = 0;
    FOREACH (database, configState.databases)
    {
        if (i == n)
        {
            name.Write(database->name);
            return;
        }
            
        i++;
    }
}

unsigned Client::GetNumTables(uint64_t databaseID)
{
    ConfigDatabase*     database;
    
    GET_CONFIG_STATE_OR_RETURN(0);

    database = NULL;
    FOREACH (database, configState.databases)
    {
        if (database->databaseID == databaseID)
            break;
    }

    if (!database)
        return 0;
    
    return database->tables.GetLength();
}

uint64_t Client::GetTableIDAt(uint64_t databaseID, unsigned n)
{
    ConfigDatabase* database;
    uint64_t*       itTableID;
    unsigned        i;

    GET_CONFIG_STATE_OR_RETURN(0);
    
    database = NULL;
    FOREACH (database, configState.databases)
    {
        if (database->databaseID == databaseID)
            break;
    }

    if (!database)
        return 0;

    i = 0;
    FOREACH (itTableID, database->tables)
    {
        if (i == n)
            return *itTableID;
        i++;
    }
    
    return 0;
}

void Client::GetTableNameAt(uint64_t databaseID, unsigned n, Buffer& name)
{
    ConfigDatabase* database;
    ConfigTable*    table;
    uint64_t*       itTableID;
    unsigned        i;

    GET_CONFIG_STATE_OR_RETURN();
    
    database = NULL;
    FOREACH (database, configState.databases)
    {
        if (database->databaseID == databaseID)
            break;
    }

    if (!database)
        return;
    
    i = 0;
    FOREACH (itTableID, database->tables)
    {
        if (i == n)
            break;
        i++;
    }
    
    if (!itTableID)
        return;
    
    FOREACH (table, configState.tables)
    {
        if (table->tableID == *itTableID)
        {
            name.Write(table->name);
            return;
        }
    }
}    

int Client::Get(uint64_t tableID, const ReadBuffer& key)
{
    Request*    req;
    Request*    proxiedRequest;

    if (key.GetLength() == 0)
        return SDBP_API_ERROR;
    
    VALIDATE_CONTROLLERS();
    CLIENT_MUTEX_GUARD_DECLARE();
    
    req = new Request;
    req->Get(NextCommandID(), configState.paxosID, tableID, (ReadBuffer&) key);

    // find
    proxiedRequest = proxy.Find(req);
    if (proxiedRequest)
    {
        delete req;
        if (proxiedRequest->type == CLIENTREQUEST_SET)
        {
            result->proxied = true;
            result->proxiedValue.Wrap(proxiedRequest->value);
            return SDBP_SUCCESS;
        }
        else if (proxiedRequest->type == CLIENTREQUEST_DELETE)
        {
            return SDBP_FAILED;
        }
        else
            ASSERT_FAIL();
    }
        
    AppendDataRequest(req);

    CLIENT_MUTEX_GUARD_UNLOCK();
    EventLoop();
    return result->GetCommandStatus();
}

int Client::Set(uint64_t tableID, const ReadBuffer& key, const ReadBuffer& value)
{
    Request* req;

    req = new Request;
    req->Set(NextCommandID(), configState.paxosID,
     tableID, (ReadBuffer&) key, (ReadBuffer&) value);

    return ProxiedRequest(req);
}

int Client::Delete(uint64_t tableID, const ReadBuffer& key)
{
    Request* req;

    req = new Request;
    req->Delete(NextCommandID(), configState.paxosID,
     tableID, (ReadBuffer&) key);

    return ProxiedRequest(req);
}

int Client::Add(uint64_t tableID, const ReadBuffer& key, int64_t number)
{
    int         status;
    Request*    req;
    ReadBuffer  value;
    Buffer      newValue;
    int64_t     oldNumber;
    int64_t     newNumber;

    if (InTransaction())
    {
        status = Get(tableID, key);
        if (status == SDBP_FAILED) // dne
        {
            oldNumber = 0;
        }
        else
        {
            if (status != SDBP_SUCCESS)
                return status;
            status = result->GetValue(value);
            if (status != SDBP_SUCCESS)
                return status;
            status = value.Readf("%I", &oldNumber);
            if (status <= 0)
                return SDBP_FAILED;
        }
        newNumber = oldNumber + number;
        newValue.Writef("%I", newNumber);
        status = Set(tableID, key, newValue);
        return status;
    }

    req = new Request;
    req->Add(NextCommandID(), configState.paxosID, tableID, (ReadBuffer&) key, number);

    return PassthroughRequest(req);
}

int Client::SequenceSet(uint64_t tableID, const ReadBuffer& key, const uint64_t value)
{
    Request*    req;

    req = new Request;<commit_msg>Fix bug where ADD failed if inside transaction and key did not previously exist.<commit_after>#include "SDBPClient.h"
#include "SDBPControllerConnection.h"
#include "SDBPController.h"
#include "SDBPShardConnection.h"
#include "SDBPPooledShardConnection.h"
#include "SDBPClientConsts.h"
#include "System/Common.h"
#include "System/Macros.h"
#include "System/IO/IOProcessor.h"
#include "System/Events/EventLoop.h"
#include "System/Threading/ThreadPool.h"
#include "System/Threading/LockGuard.h"
#include "Framework/Replication/PaxosLease/PaxosLease.h"
#include "Application/Common/ClientRequest.h"
#include "Application/Common/ClientResponse.h"

// TODO: find out the optimal size
#define MAX_IO_CONNECTION               32768
#define DEFAULT_BATCH_LIMIT             (1*MB)

#ifndef CLIENT_MULTITHREADED
#define CLIENT_MULTITHREADED
#endif

#ifdef CLIENT_MULTITHREADED

// globalMutex protects the underlying single threaded IO and Event handling layer
Mutex                                   globalMutex;                    
unsigned                                numClients = 0;
ThreadPool*                             ioThread = NULL;
Signal                                  ioThreadSignal;

#define CLIENT_MUTEX_GUARD_DECLARE()    MutexGuard clientMutexGuard(mutex)
#define CLIENT_MUTEX_GUARD_LOCK()       clientMutexGuard.Lock()
#define CLIENT_MUTEX_GUARD_UNLOCK()     clientMutexGuard.Unlock()

#define CLIENT_MUTEX_LOCK()             Lock()
#define CLIENT_MUTEX_UNLOCK()           Unlock()

#define GLOBAL_MUTEX_GUARD_DECLARE()    MutexGuard globalMutexGuard(globalMutex)
#define GLOBAL_MUTEX_GUARD_LOCK()       globalMutexGuard.Lock()
#define GLOBAL_MUTEX_GUARD_UNLOCK()     globalMutexGuard.Unlock()

#define GLOBAL_MUTEX_LOCK()             globalMutex.Lock()
#define GLOBAL_MUTEX_UNLOCK()           globalMutex.Unlock()

#else // CLIENT_MULTITHREADED

#define CLIENT_MUTEX_GUARD_DECLARE()
#define CLIENT_MUTEX_GUARD_LOCK()
#define CLIENT_MUTEX_GUARD_UNLOCK()

#define CLIENT_MUTEX_LOCK()
#define CLIENT_MUTEX_UNLOCK()

#define GLOBAL_MUTEX_GUARD_DECLARE()
#define GLOBAL_MUTEX_GUARD_LOCK()
#define GLOBAL_MUTEX_GUARD_UNLOCK()

#define GLOBAL_MUTEX_LOCK()
#define GLOBAL_MUTEX_UNLOCK()

#endif // CLIENT_MULTITHREADED

#define VALIDATE_CONTROLLERS()       \
    if (controller == 0)             \
        return SDBP_API_ERROR;

#define VALIDATE_CONFIG_STATE()     \
    if (configState == NULL)        \
        return SDBP_API_ERROR;

using namespace SDBPClient;

static inline uint64_t Hash(uint64_t h)
{
    return h;
}

static inline uint64_t Key(const ShardConnection* conn)
{
    return conn->GetNodeID();
}

static inline int KeyCmp(uint64_t a, uint64_t b)
{
    if (a < b)
        return -1;
    if (a > b)
        return 1;
    return 0;
}

Client::Client()
{
    controller = NULL;
    globalMutex.SetName("ClientGlobalMutex");
    mutexName.Writef("Client_%p", this);
    mutexName.NullTerminate();
    mutex.SetName(mutexName.GetBuffer());
}

Client::~Client()
{
    Shutdown();
}

int Client::Init(int nodec, const char* nodev[])
{
    // sanity check on parameters
    if (nodec <= 0 || nodev == NULL)
        return SDBP_API_ERROR;

    // set up IO thread, IOProcessor and EventLoop
    GLOBAL_MUTEX_GUARD_DECLARE();
    if (!IOProcessor::Init(MAX_IO_CONNECTION))
        return SDBP_API_ERROR;

    SeedRandomWith(ThreadPool::GetThreadID());

    if (numClients == 0)
    {
        Log_Debug("Creating IOThread");
        ioThreadSignal.SetWaiting(true);
        ioThread = ThreadPool::Create(1);
        ioThread->Execute(MFUNC(Client, IOThreadFunc));
        ioThread->Start();
        ioThreadSignal.Wait();
        Log_Debug("IOThread started");
    }
    numClients++;
    Log_Debug("Number of clients: %u", numClients);
    GLOBAL_MUTEX_GUARD_UNLOCK();

    IOProcessor::BlockSignals(IOPROCESSOR_BLOCK_INTERACTIVE);

    // set default timeouts
    masterTimeout.SetCallable(MFUNC(Client, OnMasterTimeout));
    masterTimeout.SetDelay(3 * PAXOSLEASE_MAX_LEASE_TIME);
    globalTimeout.SetCallable(MFUNC(Client, OnGlobalTimeout));
    globalTimeout.SetDelay(SDBP_DEFAULT_TIMEOUT);

    // set defaults
    commandID = 0;
    result = NULL;
    batchMode = SDBP_BATCH_DEFAULT;
    batchLimit = DEFAULT_BATCH_LIMIT;
    proxy.Init();
    consistencyMode = SDBP_CONSISTENCY_STRICT;
    connectivityStatus = SDBP_NOCONNECTION;
    timeoutStatus = SDBP_SUCCESS;
    transactionStatus = SDBP_SUCCESS;
    numControllerRequests = 0;
    numNestedTransactions = 0;
    transactionQuorumID = 0;
    next = prev = this;

    // create the first result object
    result = new Result;

    // wait for starting properly
    controller = Controller::GetController(this, nodec, nodev);
    if (!controller)
    {
        Shutdown();
        return SDBP_API_ERROR;
    }
    Lock();
    controller->AddClient(this);
    Unlock();

    return SDBP_SUCCESS;
}

void Client::Shutdown()
{
    if (controller == NULL)
        return;
    
    if (batchMode != SDBP_BATCH_NOAUTOSUBMIT)
        Submit();

    isShutdown.SetWaiting(true);
    if (EventLoop::IsRunning())
    {
        onClientShutdown.SetCallable(MFUNC(Client, OnClientShutdown));
        EventLoop::Add(&onClientShutdown);
        isShutdown.Wait();
    }
    EventLoop::Remove(&onClientShutdown);

    GLOBAL_MUTEX_GUARD_DECLARE();
    ASSERT(numClients != 0);
    numClients--;
    if (numClients == 0)
    {
        PooledShardConnection::ShutdownPool();
        
        Log_Debug("Stopping IOThread");
        EventLoop::Stop();
        ioThread->WaitStop();
        delete ioThread;
        ioThread = NULL;
        Log_Debug("IOThread destroyed");
    }

    IOProcessor::Shutdown();

    GLOBAL_MUTEX_GUARD_UNLOCK();

    delete result;
}

// this is always called from the IO thread with the client lock locked
void Client::OnClientShutdown()
{
    ASSERT(numNestedTransactions >= 0);
    if (numNestedTransactions > 0)
        Log_Trace("numNestedTransactions = %d", numNestedTransactions);

    numNestedTransactions = 0;
    ReleaseShardConnections();
    
    EventLoop::Remove(&masterTimeout);
    EventLoop::Remove(&globalTimeout);

    if (!controller->IsShuttingDown())
        Controller::CloseController(this, controller);
    controller = NULL;
    
    shardConnections.DeleteTree();

    DeleteQuorumRequests();
    quorumRequests.Clear();
    numControllerRequests = 0;

    isShutdown.Wake();
}

void Client::SetBatchLimit(unsigned batchLimit_)
{
    batchLimit = batchLimit_;
}

void Client::SetConsistencyMode(int consistencyMode_)
{
    consistencyMode = consistencyMode_;
}

void Client::SetBatchMode(int batchMode_)
{
    batchMode = batchMode_;
}

void Client::SetGlobalTimeout(uint64_t timeout)
{
    if (globalTimeout.IsActive())
    {
        EventLoop::Remove(&globalTimeout);
        globalTimeout.SetDelay(timeout);
        EventLoop::Add(&globalTimeout);
    }
    else
        globalTimeout.SetDelay(timeout);
}

void Client::SetMasterTimeout(uint64_t timeout)
{
    Log_Trace("timeout: %U", timeout);

    if (masterTimeout.IsActive())
    {
        EventLoop::Remove(&masterTimeout);
        masterTimeout.SetDelay(timeout);
        EventLoop::Add(&masterTimeout);
    }
    else
        masterTimeout.SetDelay(timeout);
}

uint64_t Client::GetGlobalTimeout()
{
    return globalTimeout.GetDelay();
}

uint64_t Client::GetMasterTimeout()
{
    return masterTimeout.GetDelay();
}

// this must be called with the client lock locked!
ConfigState* Client::UnprotectedGetConfigState()
{
    if (!configState.hasMaster)
    {
        result->Close();
        Unlock();
        EventLoop();
        Lock();
    }

    return &configState;
}

ConfigState* Client::GetConfigState()
{
    Log_Trace();

    CLIENT_MUTEX_GUARD_DECLARE();

    if (controller == NULL)
        return NULL;
    
    if (!configState.hasMaster)
    {
        result->Close();
        CLIENT_MUTEX_GUARD_UNLOCK();
        EventLoop();
        CLIENT_MUTEX_GUARD_LOCK();
    }

    return &configState;
}

void Client::CloneConfigState(ConfigState& configState_)
{
    CLIENT_MUTEX_GUARD_DECLARE();

    if (controller == NULL)
        return;
    
    if (!configState.hasMaster)
    {
        result->Close();
        CLIENT_MUTEX_GUARD_UNLOCK();
        EventLoop();
        CLIENT_MUTEX_GUARD_LOCK();
    }

    configState_ = configState;
}

void Client::WaitConfigState()
{
    CLIENT_MUTEX_GUARD_DECLARE();

    if (controller == NULL)
        return;

    // delete configState, so EventLoop() will wait for a new one
    configState.hasMaster = false;
    result->Close();
    CLIENT_MUTEX_GUARD_UNLOCK();
    EventLoop();
    CLIENT_MUTEX_GUARD_LOCK();
}

Result* Client::GetResult()
{
    CLIENT_MUTEX_GUARD_DECLARE();

    if (controller == NULL)
        return NULL;

    Result* tmp;
    
    tmp = result;
    result = new Result;
    return tmp;
}

int Client::GetTransportStatus()
{
    VALIDATE_CONTROLLERS();
    return result->GetTransportStatus();
}

int Client::GetConnectivityStatus()
{
    return connectivityStatus;
}

int Client::GetTimeoutStatus()
{
    return timeoutStatus;
}

int Client::GetCommandStatus()
{
    VALIDATE_CONTROLLERS();
    return result->GetCommandStatus();
}

int Client::CreateDatabase(ReadBuffer& name)
{
    Request* req;

    req = new Request;
    req->CreateDatabase(controller->NextCommandID(), name);

    return ConfigRequest(req);
}

int Client::RenameDatabase(uint64_t databaseID, const ReadBuffer& name)
{
    Request* req;

    req = new Request;
    req->RenameDatabase(controller->NextCommandID(), databaseID, (ReadBuffer&) name);

    return ConfigRequest(req);
}

int Client::DeleteDatabase(uint64_t databaseID)
{
    Request* req;

    req = new Request;
    req->DeleteDatabase(controller->NextCommandID(), databaseID);

    return ConfigRequest(req);
}

int Client::CreateTable(uint64_t databaseID, uint64_t quorumID, ReadBuffer& name)
{
    Request* req;

    req = new Request;
    req->CreateTable(controller->NextCommandID(), databaseID, quorumID, name);

    return ConfigRequest(req);
}

int Client::RenameTable(uint64_t tableID, ReadBuffer& name)
{
    Request* req;

    req = new Request;
    req->RenameTable(controller->NextCommandID(), tableID, name);

    return ConfigRequest(req);
}

int Client::DeleteTable(uint64_t tableID)
{
    Request* req;

    req = new Request;
    req->DeleteTable(controller->NextCommandID(), tableID);

    return ConfigRequest(req);
}

int Client::TruncateTable(uint64_t tableID)
{
    Request* req;

    req = new Request;
    req->TruncateTable(controller->NextCommandID(), tableID);

    return ConfigRequest(req);
}

#define GET_CONFIG_STATE_OR_RETURN(...) \
    CLIENT_MUTEX_GUARD_DECLARE();       \
                                        \
    if (controller == NULL)             \
        return __VA_ARGS__;             \
                                        \
    if (!configState.hasMaster)         \
    {                                   \
        result->Close();                \
        CLIENT_MUTEX_GUARD_UNLOCK();    \
        EventLoop();                    \
        CLIENT_MUTEX_GUARD_LOCK();      \
    }                                   \
    if (!configState.hasMaster)         \
        return __VA_ARGS__;

unsigned Client::GetNumQuorums()
{
    GET_CONFIG_STATE_OR_RETURN(0);
    
    return configState.quorums.GetLength();
}

uint64_t Client::GetQuorumIDAt(unsigned n)
{
    ConfigQuorum*   quorum;
    unsigned        i;

    GET_CONFIG_STATE_OR_RETURN(0);
    
    i = 0;
    FOREACH (quorum, configState.quorums)
    {
        if (i == n)
            return quorum->quorumID;
            
        i++;
    }

    return 0;
}

void Client::GetQuorumNameAt(unsigned n, Buffer& name)
{
    ConfigQuorum*   quorum;
    unsigned        i;

    GET_CONFIG_STATE_OR_RETURN();
    
    i = 0;
    FOREACH (quorum, configState.quorums)
    {
        if (i == n)
        {
            name.Write(quorum->name);
            return;
        }
            
        i++;
    }
}

unsigned Client::GetNumDatabases()
{
    GET_CONFIG_STATE_OR_RETURN(0);
    
    return configState.databases.GetLength();
}

uint64_t Client::GetDatabaseIDAt(unsigned n)
{
    ConfigDatabase* database;
    unsigned        i;

    GET_CONFIG_STATE_OR_RETURN(0);
    
    i = 0;
    FOREACH (database, configState.databases)
    {
        if (i == n)
            return database->databaseID;
            
        i++;
    }

    return 0;
}

void Client::GetDatabaseNameAt(unsigned n, Buffer& name)
{
    ConfigDatabase* database;
    unsigned        i;

    GET_CONFIG_STATE_OR_RETURN();

    i = 0;
    FOREACH (database, configState.databases)
    {
        if (i == n)
        {
            name.Write(database->name);
            return;
        }
            
        i++;
    }
}

unsigned Client::GetNumTables(uint64_t databaseID)
{
    ConfigDatabase*     database;
    
    GET_CONFIG_STATE_OR_RETURN(0);

    database = NULL;
    FOREACH (database, configState.databases)
    {
        if (database->databaseID == databaseID)
            break;
    }

    if (!database)
        return 0;
    
    return database->tables.GetLength();
}

uint64_t Client::GetTableIDAt(uint64_t databaseID, unsigned n)
{
    ConfigDatabase* database;
    uint64_t*       itTableID;
    unsigned        i;

    GET_CONFIG_STATE_OR_RETURN(0);
    
    database = NULL;
    FOREACH (database, configState.databases)
    {
        if (database->databaseID == databaseID)
            break;
    }

    if (!database)
        return 0;

    i = 0;
    FOREACH (itTableID, database->tables)
    {
        if (i == n)
            return *itTableID;
        i++;
    }
    
    return 0;
}

void Client::GetTableNameAt(uint64_t databaseID, unsigned n, Buffer& name)
{
    ConfigDatabase* database;
    ConfigTable*    table;
    uint64_t*       itTableID;
    unsigned        i;

    GET_CONFIG_STATE_OR_RETURN();
    
    database = NULL;
    FOREACH (database, configState.databases)
    {
        if (database->databaseID == databaseID)
            break;
    }

    if (!database)
        return;
    
    i = 0;
    FOREACH (itTableID, database->tables)
    {
        if (i == n)
            break;
        i++;
    }
    
    if (!itTableID)
        return;
    
    FOREACH (table, configState.tables)
    {
        if (table->tableID == *itTableID)
        {
            name.Write(table->name);
            return;
        }
    }
}    

int Client::Get(uint64_t tableID, const ReadBuffer& key)
{
    Request*    req;
    Request*    proxiedRequest;

    if (key.GetLength() == 0)
        return SDBP_API_ERROR;
    
    VALIDATE_CONTROLLERS();
    CLIENT_MUTEX_GUARD_DECLARE();
    
    req = new Request;
    req->Get(NextCommandID(), configState.paxosID, tableID, (ReadBuffer&) key);

    // find
    proxiedRequest = proxy.Find(req);
    if (proxiedRequest)
    {
        delete req;
        if (proxiedRequest->type == CLIENTREQUEST_SET)
        {
            result->proxied = true;
            result->proxiedValue.Wrap(proxiedRequest->value);
            return SDBP_SUCCESS;
        }
        else if (proxiedRequest->type == CLIENTREQUEST_DELETE)
        {
            return SDBP_FAILED;
        }
        else
            ASSERT_FAIL();
    }
        
    AppendDataRequest(req);

    CLIENT_MUTEX_GUARD_UNLOCK();
    EventLoop();
    return result->GetCommandStatus();
}

int Client::Set(uint64_t tableID, const ReadBuffer& key, const ReadBuffer& value)
{
    Request* req;

    req = new Request;
    req->Set(NextCommandID(), configState.paxosID,
     tableID, (ReadBuffer&) key, (ReadBuffer&) value);

    return ProxiedRequest(req);
}

int Client::Delete(uint64_t tableID, const ReadBuffer& key)
{
    Request* req;

    req = new Request;
    req->Delete(NextCommandID(), configState.paxosID,
     tableID, (ReadBuffer&) key);

    return ProxiedRequest(req);
}

int Client::Add(uint64_t tableID, const ReadBuffer& key, int64_t number)
{
    int         status;
    Request*    req;
    ReadBuffer  value;
    Buffer      newValue;
    int64_t     oldNumber;
    int64_t     newNumber;

    if (InTransaction())
    {
        status = Get(tableID, key);
        if (status == SDBP_FAILED)
        {
            // key does not exist, default is 0
            oldNumber = 0;
        }
        else
        {
            if (status != SDBP_SUCCESS)
                return status;
            status = result->GetValue(value);
            if (status != SDBP_SUCCESS)
                return status;
            status = value.Readf("%I", &oldNumber);
            if (status <= 0)
                return SDBP_FAILED;
        }
        newNumber = oldNumber + number;
        newValue.Writef("%I", newNumber);
        status = Set(tableID, key, newValue);
        return status;
    }

    req = new Request;
    req->Add(NextCommandID(), configState.paxosID, tableID, (ReadBuffer&) key, number);

    return PassthroughRequest(req);
}

int Client::SequenceSet(uint64_t tableID, const ReadBuffer& key, const uint64_t value)
{
    Request*    req;

    req = new Request;
{'fraction_non_alphanumeric': 0.10809540594524733, 'fraction_numerical': 0.006550346610269064, 'mean_word_length': 2.5750288774545838, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 109, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6026881', 'n_tokens_mistral': 11129, 'n_tokens_neox': 10712, 'n_words': 2096}