server.cpp

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

{
    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->moveNext()))) 
    { 
        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->movePrev()))) 
    { 
        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::tpRectangle:
            p = pack_rectangle(p, (cli_rectangle_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::remove_current(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->remove();
            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) { 
                    len = cb->len;
                    break;
                }
            }
            offs = DOALIGN(offs, fd->components->alignment)
                + len*fd->components->dbsSize;
        }
    } while ((fd = fd->next) != first);

    old_data = new char[rec->size];