Scale Stack 0.3 Developer Documentation

scalestack/database_proxy/client_service.cc

Go to the documentation of this file.

/*
 * Scale Stack
 *
 * Copyright 2010 Eric Day
 *
 * 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.
 */

#include "config.h"

#include <algorithm>

#include <scalestack/database_proxy/client.h>
#include <scalestack/database_proxy/client_service.h>
#include <scalestack/database_proxy/connection.h>
#include <scalestack/kernel/module.h>
#include <scalestack/kernel/option.h>

using namespace std;

namespace scalestack
{
namespace database_proxy
{

/*
 * Public functions.
 */

client_service::client_service(kernel::module& module):
  network::stream_service(module),
  _max_clients(_module.get_option("max_clients").get_size_value()),
  _current_clients(),
  _queue_mutex(),
  _waiting_connections(),
  _free_clients(),
  _connection_service(module, *this)
{
  int return_code = pthread_mutex_init(&_queue_mutex, NULL);
  if (return_code != 0)
    _module.log_fatal("Failed pthread_mutex_init: %d", return_code);
}

client_service::~client_service()
{
  int return_code = pthread_mutex_destroy(&_queue_mutex);
  if (return_code != 0)
    _module.log_error("Failed pthread_mutex_destroy: %d", return_code);
}

network::stream* client_service::add_stream(void)
{
  return new client(*this);
}

void client_service::remove_stream(network::stream* old_stream)
{
  pthread_mutex_lock(&_queue_mutex);
  --_current_clients;
  pthread_mutex_unlock(&_queue_mutex);
  delete old_stream;
}

client* client_service::get_client(connection* bound_connection, bool need_handshake)
{
  pthread_mutex_lock(&_queue_mutex);

  client* new_client;
  vector<connection*>::reverse_iterator found_connection;
  found_connection = find(_waiting_connections.rbegin(),
                          _waiting_connections.rend(),
                          bound_connection);

  if (_free_clients.empty() ||
      (_current_clients != _max_clients && need_handshake))
  {
    if (_current_clients == _max_clients)
    {
      if (found_connection == _waiting_connections.rend())
        _waiting_connections.insert(_waiting_connections.begin(), bound_connection);
      pthread_mutex_unlock(&_queue_mutex);
      return NULL;
    }

    new_client = static_cast<client*>(_connection_service.add_connection(_module.get_option("client_address").get_value()));
    ++_current_clients;
  }
  else
  {
    new_client = _free_clients.back();
    _free_clients.pop_back();
  }

  if (found_connection != _waiting_connections.rend())
    _waiting_connections.erase(found_connection.base() - 1);

  pthread_mutex_unlock(&_queue_mutex);

  new_client->set_connection(bound_connection, need_handshake);
  return new_client;
}

void client_service::yield_client(client* bound_client)
{
  pthread_mutex_lock(&_queue_mutex);

  _free_clients.push_back(bound_client);
  if (!_waiting_connections.empty())
    _waiting_connections.back()->run_now();

  pthread_mutex_unlock(&_queue_mutex);
}

void client_service::remove_from_queue(connection* bound_connection)
{
  pthread_mutex_lock(&_queue_mutex);

  vector<connection*>::iterator found_connection;
  found_connection = find(_waiting_connections.begin(),
                          _waiting_connections.end(),
                          bound_connection);
  if (found_connection != _waiting_connections.end())
    _waiting_connections.erase(found_connection);

  pthread_mutex_unlock(&_queue_mutex);
}

} /* namespace database_proxy */
} /* namespace scalestack */

---