server.cpp

FastDb是高效的内存数据库系统

bool dbServer::unfreeze(dbSession* session, int stmt_id)
{
  dbStatement* stmt = findStatement(session, stmt_id);
  int4 response = cli_ok;

  if (stmt == NULL || stmt->cursor == NULL)
  {
    response = cli_bad_descriptor;
  }
  else
  {
    stmt->cursor->unfreeze();
  }

  pack4(response);
  return session->sock->write(&response, sizeof response);
}

bool dbServer::get_first(dbSession* session, int stmt_id)
{
  dbStatement* stmt = findStatement(session, stmt_id);
  int4 response;

  if (stmt == NULL || stmt->cursor == NULL)
  {
    response = cli_bad_descriptor;
  }
  else if (!stmt->cursor->gotoFirst())
  {
    response = cli_not_found;
  }
  else
  {
    return fetch(session, stmt);
  }

  pack4(response);
  return session->sock->write(&response, sizeof response);
}

bool dbServer::get_last(dbSession* session, int stmt_id)
{
  dbStatement* stmt = findStatement(session, stmt_id);
  int4 response;

  if (stmt == NULL || stmt->cursor == NULL)
  {
    response = cli_bad_descriptor;
  }
  else if (!stmt->cursor->gotoLast())
  {
    response = cli_not_found;
  }
  else
  {
    return fetch(session, stmt);
  }

  pack4(response);
  return session->sock->write(&response, sizeof response);
}

bool dbServer::get_next(dbSession* session, int stmt_id)
{
  dbStatement* stmt = findStatement(session, stmt_id);
  int4 response;

  if (stmt == NULL || stmt->cursor == NULL)
  {
    response = cli_bad_descriptor;
  }
  else if (!((stmt->firstFetch && stmt->cursor->gotoFirst()) ||
             (!stmt->firstFetch && stmt->cursor->gotoNext())))
  {
    response = cli_not_found;
  }
  else
  {
    return fetch(session, stmt);
  }

  pack4(response);
  return session->sock->write(&response, sizeof response);
}

bool dbServer::get_prev(dbSession* session, int stmt_id)
{
  dbStatement* stmt = findStatement(session, stmt_id);
  int4 response;

  if (stmt == NULL || stmt->cursor == NULL)
  {
    response = cli_bad_descriptor;
  }
  else if (!((stmt->firstFetch && stmt->cursor->gotoLast()) ||
             (!stmt->firstFetch && stmt->cursor->gotoPrev())))
  {
    response = cli_not_found;
  }
  else
  {
    return fetch(session, stmt);
  }

  pack4(response);
  return session->sock->write(&response, sizeof response);
}


bool dbServer::skip(dbSession* session, int stmt_id, char* buf)
{
  dbStatement* stmt = findStatement(session, stmt_id);
  int4 response;

  if (stmt == NULL || stmt->cursor == NULL)
  {
    response = cli_bad_descriptor;
  }
  else
  {
    int n = unpack4(buf);

    if ((n > 0 && !((stmt->firstFetch && stmt->cursor->gotoFirst() && stmt->cursor->skip(n-1)
                     || (!stmt->firstFetch && stmt->cursor->skip(n)))))
        || (n < 0 && !((stmt->firstFetch && stmt->cursor->gotoLast() && stmt->cursor->skip(n+1)
                        || (!stmt->firstFetch && stmt->cursor->skip(n))))))
    {
      response = cli_not_found;
    }
    else
    {
      return fetch(session, stmt);
    }
  }

  pack4(response);
  return session->sock->write(&response, sizeof response);
}

bool dbServer::seek(dbSession* session, int stmt_id, char* buf)
{
  dbStatement* stmt = findStatement(session, stmt_id);
  int4 response;

  if (stmt == NULL || stmt->cursor == NULL)
  {
    response = cli_bad_descriptor;
  }
  else
  {
    oid_t oid = unpack_oid(buf);
    int pos = stmt->cursor->seek(oid);

    if (pos < 0)
    {
      response = cli_not_found;
    }
    else
    {
      return fetch(session, stmt, pos);
    }
  }

  pack4(response);
  return session->sock->write(&response, sizeof response);
}

bool dbServer::fetch(dbSession* session, dbStatement* stmt, oid_t result)
{
  int4 response;
  dbColumnBinding* cb;

  stmt->firstFetch = false;

  if (stmt->cursor->isEmpty())
  {
    response = cli_not_found;
    pack4(response);
    return session->sock->write(&response, sizeof response);
  }

  int msg_size = sizeof(cli_oid_t) + 4;
  char* data = (char*)db->getRow(stmt->cursor->currId);

  for (cb = stmt->columns; cb != NULL; cb = cb->next)
  {
    if (cb->cliType == cli_autoincrement)
    {
      msg_size += 4;
    }
    else if (cb->cliType >= cli_array_of_oid)
    {
      msg_size += 4 + ((dbVarying*)(data + cb->fd->dbsOffs))->size
                  * sizeof_type[cb->cliType - cli_array_of_oid];
    }
    else if (cb->cliType >= cli_asciiz)
    {
      msg_size += 4 + ((dbVarying*)(data + cb->fd->dbsOffs))->size;
    }
    else
    {
      msg_size += sizeof_type[cb->cliType];
    }
  }

  if (stmt->buf_size < msg_size)
  {
    delete[] stmt->buf;
    stmt->buf = new char[msg_size];
    stmt->buf_size = msg_size;
  }

  char* p = stmt->buf;
  p = pack4(p, msg_size);
  p = pack_oid(p, result);

  for (cb = stmt->columns; cb != NULL; cb = cb->next)
  {
    char* src = data + cb->fd->dbsOffs;

    switch (cb->fd->type)
    {

    case dbField::tpBool:

    case dbField::tpInt1:

      switch (sizeof_type[cb->cliType])
      {

      case 1:
        *p++ = *src;
        break;

      case 2:
        p = pack2(p, (int2)*(char*)src);
        break;

      case 4:
        p = pack4(p, (int4)*(char*)src);
        break;

      case 8:
        p = pack8(p, (db_int8)*(char*)src);
        break;

      default:
        assert(false);
      }

      break;

    case dbField::tpInt2:

      switch (sizeof_type[cb->cliType])
      {

      case 1:
        *p++ = (char)*(int2*)src;
        break;

      case 2:
        p = pack2(p, src);
        break;

      case 4:
        p = pack4(p, (int4)*(int2*)src);
        break;

      case 8:
        p = pack8(p, (db_int8)*(int2*)src);
        break;

      default:
        assert(false);
      }

      break;

    case dbField::tpInt4:

      switch (sizeof_type[cb->cliType])
      {

      case 1:
        *p++ = (char)*(int4*)src;
        break;

      case 2:
        p = pack2(p, (int2)*(int4*)src);
        break;

      case 4:
        p = pack4(p, src);
        break;

      case 8:
        p = pack8(p, (db_int8)*(int4*)src);
        break;

      default:
        assert(false);
      }

      break;

    case dbField::tpInt8:

      switch (sizeof_type[cb->cliType])
      {

      case 1:
        *p++ = (char)*(db_int8*)src;
        break;

      case 2:
        p = pack2(p, (int2)*(db_int8*)src);
        break;

      case 4:
        p = pack4(p, (int4)*(db_int8*)src);
        break;

      case 8:
        p = pack8(p, src);
        break;

      default:
        assert(false);
      }

      break;

    case dbField::tpReal4:

      switch (cb->cliType)
      {

      case cli_real4:
        p = pack4(p, src);
        break;

      case cli_real8:
        {
          real8 temp = *(real4*)src;
          p = pack8(p, (char*)&temp);
        }

        break;

      default:
        assert(false);
      }

      break;

    case dbField::tpReal8:

      switch (cb->cliType)
      {

      case cli_real4:
        {
          real4 temp = (real4)*(real8*)src;
          p = pack4(p, (char*)&temp);
        }

        break;

      case cli_real8:
        p = pack8(p, src);
        break;

      default:
        assert(false);
      }

      break;

    case dbField::tpString:
      {
        dbVarying* v = (dbVarying*)src;
        p = pack4(p, v->size);
        memcpy(p, data + v->offs, v->size);
        p += v->size;
      }

      break;

    case dbField::tpReference:
      p = pack_oid(p, *(oid_t*)src);
      break;

    case dbField::tpArray:
      {
        dbVarying* v = (dbVarying*)src;
        int n = v->size;
        p = pack4(p, n);
        src = data + v->offs;

        switch (sizeof_type[cb->cliType-cli_array_of_oid])
        {

        case 2:

          while (--n >= 0)
          {
            p = pack2(p, src);
            src += 2;
          }

          break;

        case 4:

          while (--n >= 0)
          {
            p = pack4(p, src);
            src += 4;
          }

          break;

        case 8:

          while (--n >= 0)
          {
            p = pack8(p, src);
            src += 8;
          }

          break;

        default:
          memcpy(p, src, n);
          p += n;
        }

        break;
      }

    default:
      assert(false);
    }
  }

  assert(p - stmt->buf == msg_size);
  return session->sock->write(stmt->buf, msg_size);
}

bool dbServer::remove
  (dbSession* session, int stmt_id)
{
  dbStatement* stmt = findStatement(session, stmt_id);
  int4 response;

  if (stmt == NULL)
  {
    response = cli_bad_descriptor;
  }
  else
  {
    if (stmt->cursor->isEmpty())
    {
      response = cli_not_found;
    }
    else
    {
      stmt->cursor->removeAllSelected();
      response = cli_ok;
    }
  }

  pack4(response);
  return session->sock->write(&response, sizeof response);
}

bool dbServer::update(dbSession* session, int stmt_id, char* new_data)
{
  dbStatement* stmt = findStatement(session, stmt_id);
  dbColumnBinding* cb;
  int4 response;

  if (stmt == NULL)
  {
    response = cli_bad_descriptor;
    pack4(response);
    return session->sock->write(&response, sizeof response);
  }

  if (stmt->cursor->isEmpty())
  {
    response = cli_not_found;
    pack4(response);
    return session->sock->write(&response, sizeof response);
  }

  char* old_data = stmt->buf + sizeof(cli_oid_t) + 4;

  for (cb = stmt->columns; cb != NULL; cb = cb->next)
  {
    cb->ptr = new_data;

    if (cb->cliType >= cli_asciiz)
    {
      int new_len = unpack4(new_data);
      int old_len = unpack4(old_data);
      cb->len = new_len;

      if (cb->fd->indexType & (HASHED|INDEXED)
          && memcmp(new_data, old_data, new_len+4) != 0)
      {
        cb->fd->attr |= dbFieldDescriptor::Updated;
      }

      if (cb->cliType >= cli_array_of_oid)
      {
        new_len *= sizeof_type[cb->cliType - cli_array_of_oid];
        old_len *= sizeof_type[cb->cliType - cli_array_of_oid];
      }

      new_data += 4 + new_len;
      old_data += 4 + old_len;
    }
    else
    {
      int size = sizeof_type[cb->cliType];

      if (cb->fd->indexType & (HASHED|INDEXED)
          && memcmp(new_data, old_data, size) != 0)
      {
        cb->fd->attr |= dbFieldDescriptor::Updated;
      }

      new_data += size;
      old_data += size;
    }
  }

  db->beginTransaction(dbDatabase::dbExclusiveLock);

  dbRecord* rec = db->getRow(stmt->cursor->currId);
  dbTableDescriptor* table = stmt->query.table;
  dbFieldDescriptor *first = table->columns, *fd = first;
  size_t offs = table->fixedSize;

  do
  {
    if (fd->type == dbField::tpArray || fd->type == dbField::tpString)
    {
      int len = ((dbVarying*)((char*)rec + fd->dbsOffs))->size;

      for (cb = stmt->columns; cb != NULL; cb = cb->next)
      {
        if (cb->fd == fd)
        {