OpenNebula

/* -------------------------------------------------------------------------- */

/* Copyright 2002-2016, OpenNebula Project, OpenNebula Systems                */

/*                                                                            */

/* Licensed under the Apache License, Version 2.0 (the "License"); you may    */

/* not use this file except in compliance with the License. You may obtain    */

/* a copy of the License at                                                   */

/*                                                                            */

/* http://www.apache.org/licenses/LICENSE-2.0                                 */

/*                                                                            */

/* Unless required by applicable law or agreed to in writing, software        */

/* distributed under the License is distributed on an "AS IS" BASIS,          */

/* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.   */

/* See the License for the specific language governing permissions and        */

/* limitations under the License.                                             */

/* -------------------------------------------------------------------------- */

#ifndef LOG_DB_H_

#define LOG_DB_H_

#include <string>

#include <sstream>

#include "SqlDB.h"

#include "LogDBRequest.h"

class LogDB : public SqlDB

{

public:

    LogDB(SqlDB * _db):db(_db), term(0), index(0){};

    virtual ~LogDB(){};

    void set_term(unsigned int t)

    {

        term = t;

    }

    void set_index(unsigned int i)

    {

        next_index = i;

    }

    int exec_wr(ostringstream& cmd)

    {

        int rc;

        //TODO: WRITE RECORD IN DB

        //

        LogDBRecord * lr = new LogDBRequest(next_index, term, cmd);

        next_index++;

        //LogDBManager->triger(NEW_LOG_RECORD);

        lr.wait();

        if ( lr.result == true )

        {

            rc = exec(cmd, 0, false);

        }

        else

        {

            rc = -1;

            //Nebula::Log

        }

        return rc;

    }

    // -------------------------------------------------------------------------

    // SQL interface. Use database store implementation

    // -------------------------------------------------------------------------

    char * escape_str(const string& str)

    {

        return db->escape_str(str);

    }

    void free_str(char * str)

    {

        db->free_str(str);

    }

    bool multiple_values_support()

    {

        return db->multiple_values_support();

    }

protected:

    int exec(ostringstream& cmd, Callbackable* obj, bool quiet)

    {

        return db->exec(cmd, obj, quiet);

    }

private:

    /**

     *  Pointer to the underlying DB store

     */

    SqlDB * db;

    /**

     *  Index to be used by the next logDB record

     */

    unsigned int next_index;

    /**

     *  Current term to be included in new LogDB records generated during the

     *  term.

     */

    unsigned int term;

};

#endif /*LOG_DB_H_*/

/* -------------------------------------------------------------------------- */

/* Copyright 2002-2016, OpenNebula Project, OpenNebula Systems                */

/*                                                                            */

/* Licensed under the Apache License, Version 2.0 (the "License"); you may    */

/* not use this file except in compliance with the License. You may obtain    */

/* a copy of the License at                                                   */

/*                                                                            */

/* http://www.apache.org/licenses/LICENSE-2.0                                 */

/*                                                                            */

/* Unless required by applicable law or agreed to in writing, software        */

/* distributed under the License is distributed on an "AS IS" BASIS,          */

/* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.   */

/* See the License for the specific language governing permissions and        */

/* limitations under the License.                                             */

/* -------------------------------------------------------------------------- */

#ifndef LOG_DB_MANAGER_H_

#define LOG_DB_MANAGER_H_

#include "ActionManager.h"

#include "LogDBRecord.h"

#include "ZoneServer.h"

extern "C" void * logdb_manager_loop(void *arg);

/* -------------------------------------------------------------------------- */

/* -------------------------------------------------------------------------- */

class LogDBAction : public ActionRequest

{

public:

    enum Actions

    {

        NEW_LOGDB_RECORD,

        DELETE_SERVER

    }

    LogDBAction(Actions a, LogDBRequest * r):ActionRequest(ActionRequest::USER),

        _action(a), _request(r){};

    LogDBAction(const LogDBAction& o):ActionRequest(o._type),

        _action(o._action), _request(o._request){};

    Actions action() const

    {

        return _action;

    }

    LogDBRequest * request() const

    {

        return _request;

    }

    ActionRequest * clone() const

    {

        return new LogDBAction(*this);

    }

private:

    Action _action;

    LogDBRequest * _request;

};

// -----------------------------------------------------------------------------

// -----------------------------------------------------------------------------

class LogDBManager : public ActionListener

{

private:

    class LogDBManagerThread

    {

    public:

        LogDBManagerThread(ZoneServer * z):replicate(false), zserver(z)

        {

            pthread_mutex_init(&mutex, 0);

            pthread_cond_init(&cond, 0);

        };

        virtual ~LogDBManagerThread(){};

        void do_replication();

    private:

        pthread_t thread_id;

        pthread_mutex_t mutex;

        pthread_cond_t cond;

        bool replicate;

        ZoneServer * zserver;

    }

    /**

     *  LogDB records being replicated on followers

     */

    std::map<unsigned int, LogDBRecord *> log;

    // -------------------------------------------------------------------------

    // Action Listener interface

    // -------------------------------------------------------------------------

    void finalize_action(const ActionRequest& ar)

    {

        NebulaLog::log("DBM",Log::INFO,"Stopping LogDB Manager...");

    };

    void user_action(const ActionRequest& ar);

}

#endif /*LOG_DB_H_*/

/* -------------------------------------------------------------------------- */

/* Copyright 2002-2016, OpenNebula Project, OpenNebula Systems                */

/*                                                                            */

/* Licensed under the Apache License, Version 2.0 (the "License"); you may    */

/* not use this file except in compliance with the License. You may obtain    */

/* a copy of the License at                                                   */

/*                                                                            */

/* http://www.apache.org/licenses/LICENSE-2.0                                 */

/*                                                                            */

/* Unless required by applicable law or agreed to in writing, software        */

/* distributed under the License is distributed on an "AS IS" BASIS,          */

/* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.   */

/* See the License for the specific language governing permissions and        */

/* limitations under the License.                                             */

/* -------------------------------------------------------------------------- */

#ifndef LOG_DB_REQUEST_H_

#define LOG_DB_REQUEST_H_

#include <string>

#include <sstream>

#include "SqlDB.h"

/**

 * This class represents a log entry replication request. The replication request

 * is synchronous: once it has been replicated in a majority of followers the

 * client is notified (SqlDB::exec_wr() call) and DB updated.

 */

class LogDBRequest : public SyncRequest

{

public:

    LogDBRequest(unsigned int i, unsigned int t, std::ostringstream o):

        index(i), term(t), sql(o.str()), to_commit(-1), replicas(1)

    {

        pthread_mutex_init(&mutex, 0);

    };

    virtual ~LogDBRequest(){};

    /**

     *  This function decrements the number of remaining server to replicate

     *  this entry. If it reaches 0, the client is notified

     *    @return number of replicas for this log

     */

    int replicated()

    {

        int _replicas;

        lock();

        replicas++;

        to_commit--;

        _replicas = replicas;

        if ( to_commit == 0 )

        {

            result  = true;

            timeout = false;

            notify();

        }

        unlock();

        return _replicas;

    }

private:

    pthread_mutex_t mutex;

    /**

     *  Index for this log entry

     */

    unsigned int index;

    /**

     *  Term where this log entry was generated

     */

    unsigned int term;

    /**

     *  SQL command to exec in the DB to update (INSERT, REPLACE, DROP)

     */

    std::string sql;

    /**

     *  Remaining number of servers that need to replicate this record to commit

     *  it. Initialized to ( Number_Servers - 1 ) / 2

     */

    int to_commit;

    /**

     *  Total number of replicas for this entry

     */

    int replicas;

    /**

     *  Function to lock the request

     */

    void lock()

    {

        pthread_mutex_lock(&mutex);

    };

    /**

     *  Function to unlock the request

     */

    void unlock()

    {

        pthread_mutex_unlock(&mutex);

    };

};

#endif /*LOG_DB_REQUEST_H_*/

    enum State {

        OFFLINE   = 0,

        CANDIDATE = 1,

        FOLLOWER  = 2,

        LEADER    = 3

    };

    /**

     *  @return the ID of the server

     */

    /**

     *  Initialized follower data

     *    @param last log index

     */

    void init_follower(unsigned int last)

    {

        next  = last + 1;

        match = 0;

    }

private:

    //--------------------------------------------------------------------------

    // Volatile log index variables

    //   - commit, highest log known to be commited

    //   - applied, highest log applied to DB

    //

    //---------------------------- LEADER VARIABLES ----------------------------

    //

    //   - next, next log to send to this server

    //   - match, highest log replicated in this server

    // -------------------------------------------------------------------------

    unsigned int commit;

    unsigned int applied;

    unsigned int next;

    unsigned int match;