Fix compilation errors
This commit is contained in:
+35
-89
@@ -59,77 +59,6 @@ pending_download_list_add(PendingDownloadList *list, Address addr, SHA256 hash)
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void
|
||||
removal_list_init(RemovalList *list)
|
||||
{
|
||||
list->count = 0;
|
||||
list->capacity = 0;
|
||||
list->items = NULL;
|
||||
}
|
||||
|
||||
static void
|
||||
removal_list_free(RemovalList *list)
|
||||
{
|
||||
sys_free(list->items);
|
||||
}
|
||||
|
||||
static int
|
||||
removal_list_find(RemovalList *list, SHA256 hash)
|
||||
{
|
||||
for (int i = 0; i < list->count; i++)
|
||||
if (!memcmp(&list->items[i].hash, &hash, sizeof(SHA256)))
|
||||
return i;
|
||||
return -1;
|
||||
}
|
||||
|
||||
static int
|
||||
removal_list_add(RemovalList *list, SHA256 hash, Time marked_time)
|
||||
{
|
||||
// Check if already in list
|
||||
int idx = removal_list_find(list, hash);
|
||||
if (idx >= 0) {
|
||||
// Already marked, keep the original time
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (list->count == list->capacity) {
|
||||
int new_capacity;
|
||||
if (list->capacity == 0)
|
||||
new_capacity = 8;
|
||||
else
|
||||
new_capacity = 2 * list->capacity;
|
||||
|
||||
PendingRemoval *new_items = sys_malloc(new_capacity * sizeof(PendingRemoval));
|
||||
if (new_items == NULL)
|
||||
return -1;
|
||||
|
||||
if (list->capacity > 0) {
|
||||
memcpy(new_items, list->items, list->count * sizeof(list->items[0]));
|
||||
sys_free(list->items);
|
||||
}
|
||||
|
||||
list->items = new_items;
|
||||
list->capacity = new_capacity;
|
||||
}
|
||||
|
||||
list->items[list->count++] = (PendingRemoval) { hash, marked_time };
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void
|
||||
removal_list_remove(RemovalList *list, SHA256 hash)
|
||||
{
|
||||
int idx = removal_list_find(list, hash);
|
||||
if (idx >= 0) {
|
||||
// Remove by shifting remaining items
|
||||
if (idx < list->count - 1) {
|
||||
memmove(&list->items[idx], &list->items[idx + 1],
|
||||
(list->count - idx - 1) * sizeof(list->items[0]));
|
||||
}
|
||||
list->count--;
|
||||
}
|
||||
}
|
||||
|
||||
static int chunk_store_init(ChunkStore *store, string path)
|
||||
{
|
||||
if (create_dir(path) && errno != EEXIST)
|
||||
@@ -333,8 +262,13 @@ static int process_metadata_server_sync_2(ChunkServer *state,
|
||||
if (!binary_read(&reader, &add_count, sizeof(add_count)))
|
||||
return -1;
|
||||
|
||||
HashList tmp_list;
|
||||
hash_list_init(&tmp_list);
|
||||
Time current_time = get_current_time();
|
||||
if (current_time == INVALID_TIME) {
|
||||
assert(0); // TODO
|
||||
}
|
||||
|
||||
HashSet tmp_list;
|
||||
hash_set_init(&tmp_list);
|
||||
|
||||
for (uint32_t i = 0; i < add_count; i++) {
|
||||
|
||||
@@ -345,22 +279,24 @@ static int process_metadata_server_sync_2(ChunkServer *state,
|
||||
// Elements in ms_add_list that are not held by the
|
||||
// chunk server are added to a temporary list tmp_list
|
||||
|
||||
if (chunk_store_exists(&state->store, hash) < 0) { // TODO: does it return a boolean?
|
||||
if (hash_list_insert(&tmp_list, hash) < 0) {
|
||||
if (!chunk_store_exists(&state->store, hash)) {
|
||||
if (hash_set_insert(&tmp_list, hash) < 0) {
|
||||
assert(0); // TODO
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
hash_list_remove(&state->cs_rem_list, hash);
|
||||
hash_list_remove(&state->cs_add_list, hash);
|
||||
timed_hash_set_remove(&state->cs_rem_list, hash);
|
||||
hash_set_remove(&state->cs_add_list, hash);
|
||||
}
|
||||
|
||||
if (hash_list_merge(&state->cs_rem_list, state->cs_add_list) < 0) {
|
||||
assert(0); // TODO
|
||||
for (int i = 0; i < state->cs_add_list.count; i++) {
|
||||
if (timed_hash_set_insert(&state->cs_rem_list, state->cs_add_list.items[i], current_time) < 0) {
|
||||
assert(0); // TODO
|
||||
}
|
||||
}
|
||||
|
||||
hash_list_clear(&state->cs_add_list);
|
||||
hash_set_clear(&state->cs_add_list);
|
||||
|
||||
uint32_t rem_count;
|
||||
if (!binary_read(&reader, &rem_count, sizeof(rem_count)))
|
||||
@@ -372,7 +308,7 @@ static int process_metadata_server_sync_2(ChunkServer *state,
|
||||
if (!binary_read(&reader, &hash, sizeof(hash)))
|
||||
return -1;
|
||||
|
||||
if (timed_hash_list_insert(&state->cs_rem_list, hash, current_time) < 0) {
|
||||
if (timed_hash_set_insert(&state->cs_rem_list, hash, current_time) < 0) {
|
||||
assert(0); // TODO
|
||||
}
|
||||
}
|
||||
@@ -389,12 +325,12 @@ static int process_metadata_server_sync_2(ChunkServer *state,
|
||||
uint32_t count = tmp_list.count + state->cs_lst_list.count; // TODO: overflow
|
||||
message_write(&writer, &count, sizeof(count));
|
||||
|
||||
for (uint32_t i = 0; i < tmp_list.count; i++) {
|
||||
for (int i = 0; i < tmp_list.count; i++) {
|
||||
SHA256 hash = tmp_list.items[i];
|
||||
message_write(&writer, &hash, sizeof(hash));
|
||||
}
|
||||
|
||||
for (uint32_t i = 0; i < state->cs_lst_list.count; i++) {
|
||||
for (int i = 0; i < state->cs_lst_list.count; i++) {
|
||||
SHA256 hash = state->cs_lst_list.items[i];
|
||||
message_write(&writer, &hash, sizeof(hash));
|
||||
}
|
||||
@@ -839,6 +775,12 @@ static int send_sync_message(ChunkServer *state)
|
||||
{
|
||||
assert(state->disconnect_time == INVALID_TIME);
|
||||
|
||||
int conn_idx = tcp_index_from_tag(&state->tcp, TAG_METADATA_SERVER);
|
||||
assert(conn_idx > -1);
|
||||
|
||||
ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx);
|
||||
assert(output);
|
||||
|
||||
MessageWriter writer;
|
||||
message_writer_init(&writer, output, MESSAGE_TYPE_SYNC);
|
||||
|
||||
@@ -888,7 +830,9 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts
|
||||
|
||||
state->downloading = false;
|
||||
pending_download_list_init(&state->pending_download_list);
|
||||
removal_list_init(&state->removal_list);
|
||||
hash_set_init(&state->cs_add_list);
|
||||
hash_set_init(&state->cs_lst_list);
|
||||
timed_hash_set_init(&state->cs_rem_list);
|
||||
|
||||
char tmp[1<<10];
|
||||
if (addr.len >= (int) sizeof(tmp)) {
|
||||
@@ -942,7 +886,9 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts
|
||||
int chunk_server_free(ChunkServer *state)
|
||||
{
|
||||
pending_download_list_free(&state->pending_download_list);
|
||||
removal_list_free(&state->removal_list);
|
||||
timed_hash_set_free(&state->cs_rem_list);
|
||||
hash_set_free(&state->cs_lst_list);
|
||||
hash_set_free(&state->cs_add_list);
|
||||
chunk_store_free(&state->store);
|
||||
tcp_context_free(&state->tcp);
|
||||
return 0;
|
||||
@@ -1031,12 +977,12 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled
|
||||
Time deadline = INVALID_TIME;
|
||||
|
||||
// Remove items from the remove list that got too old
|
||||
for (int i = 0; i < state->removal_list.count; i++) {
|
||||
PendingRemoval *removal = &state->removal_list.items[i];
|
||||
Time removal_time = removal->marked_time + (Time) DELETION_TIMEOUT * 1000000000;
|
||||
for (int i = 0; i < state->cs_rem_list.count; i++) {
|
||||
TimedHash *removal = &state->cs_rem_list.items[i];
|
||||
Time removal_time = removal->time + (Time) DELETION_TIMEOUT * 1000000000;
|
||||
if (removal_time < current_time) {
|
||||
if (chunk_store_remove(&state->store, removal->hash) == 0)
|
||||
*removal = state->removal_list.items[--state->removal_list.count];
|
||||
*removal = state->cs_rem_list.items[--state->cs_rem_list.count];
|
||||
} else {
|
||||
nearest_deadline(&deadline, removal_time);
|
||||
}
|
||||
|
||||
+3
-14
@@ -26,17 +26,6 @@ typedef struct {
|
||||
PendingDownload *items;
|
||||
} PendingDownloadList;
|
||||
|
||||
typedef struct {
|
||||
SHA256 hash;
|
||||
Time time;
|
||||
} TimedHash;
|
||||
|
||||
typedef struct {
|
||||
int count;
|
||||
int capacity;
|
||||
TimedHash *items;
|
||||
} TimedHashList;
|
||||
|
||||
typedef struct {
|
||||
|
||||
bool trace;
|
||||
@@ -56,13 +45,13 @@ typedef struct {
|
||||
PendingDownloadList pending_download_list;
|
||||
|
||||
// List of chunks added since the last update
|
||||
HashList cs_add_list;
|
||||
HashSet cs_add_list;
|
||||
|
||||
// List of chunks marked for removal after a timeout
|
||||
TimedHashList cs_rem_list;
|
||||
TimedHashSet cs_rem_list;
|
||||
|
||||
// List of chunks that were lost due to errors or forceful removals of chunk files
|
||||
HashList cs_lst_list;
|
||||
HashSet cs_lst_list;
|
||||
|
||||
} ChunkServer;
|
||||
|
||||
|
||||
+148
@@ -0,0 +1,148 @@
|
||||
#include <assert.h>
|
||||
#include <string.h>
|
||||
|
||||
#include "system.h"
|
||||
#include "hash_set.h"
|
||||
|
||||
void hash_set_init(HashSet *set)
|
||||
{
|
||||
set->items = NULL;
|
||||
set->count = 0;
|
||||
set->capacity = 0;
|
||||
}
|
||||
|
||||
void hash_set_free(HashSet *set)
|
||||
{
|
||||
sys_free(set->items);
|
||||
set->items = NULL;
|
||||
}
|
||||
|
||||
void hash_set_clear(HashSet *set)
|
||||
{
|
||||
free(set->items);
|
||||
set->items = NULL;
|
||||
set->count = 0;
|
||||
set->capacity = 0;
|
||||
}
|
||||
|
||||
int hash_set_insert(HashSet *set, SHA256 hash)
|
||||
{
|
||||
// Avoid duplicates
|
||||
for (int i = 0; i < set->count; i++)
|
||||
if (!memcmp(&set->items[i], &hash, sizeof(SHA256)))
|
||||
return 0; // Already present
|
||||
|
||||
if (set->count == set->capacity) {
|
||||
|
||||
int new_capacity;
|
||||
if (set->items == NULL)
|
||||
new_capacity = 16;
|
||||
else
|
||||
new_capacity = 2 * set->capacity;
|
||||
|
||||
SHA256 *new_items = sys_realloc(set->items, new_capacity * sizeof(SHA256));
|
||||
if (new_items == NULL)
|
||||
return -1;
|
||||
|
||||
set->items = new_items;
|
||||
set->capacity = new_capacity;
|
||||
}
|
||||
|
||||
set->items[set->count++] = hash;
|
||||
return 0;
|
||||
}
|
||||
|
||||
bool hash_set_remove(HashSet *set, SHA256 hash)
|
||||
{
|
||||
for (int i = 0; i < set->count; i++)
|
||||
if (!memcmp(&hash, &set->items[i], sizeof(SHA256))) {
|
||||
set->items[i] = set->items[--set->count];
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
bool hash_set_contains(HashSet *set, SHA256 hash)
|
||||
{
|
||||
for (int i = 0; i < set->count; i++)
|
||||
if (!memcmp(&hash, &set->items[i], sizeof(SHA256)))
|
||||
return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
int hash_set_merge(HashSet *dst, HashSet src)
|
||||
{
|
||||
assert(0); // TODO
|
||||
}
|
||||
|
||||
void hash_set_remove_set(HashSet *dst, HashSet src)
|
||||
{
|
||||
assert(0); // TODO
|
||||
}
|
||||
|
||||
void timed_hash_set_init(TimedHashSet *set)
|
||||
{
|
||||
set->items = NULL;
|
||||
set->count = 0;
|
||||
set->capacity = 0;
|
||||
}
|
||||
|
||||
void timed_hash_set_free(TimedHashSet *set)
|
||||
{
|
||||
sys_free(set->items);
|
||||
set->items = NULL;
|
||||
}
|
||||
|
||||
int timed_hash_set_find(TimedHashSet *set, SHA256 hash)
|
||||
{
|
||||
for (int i = 0; i < set->count; i++)
|
||||
if (!memcmp(&set->items[i].hash, &hash, sizeof(SHA256)))
|
||||
return i;
|
||||
return -1;
|
||||
}
|
||||
|
||||
int timed_hash_set_insert(TimedHashSet *set, SHA256 hash, Time time)
|
||||
{
|
||||
// Check if already in set
|
||||
int idx = timed_hash_set_find(set, hash);
|
||||
if (idx >= 0) {
|
||||
// Already marked, keep the original time
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (set->count == set->capacity) {
|
||||
int new_capacity;
|
||||
if (set->capacity == 0)
|
||||
new_capacity = 8;
|
||||
else
|
||||
new_capacity = 2 * set->capacity;
|
||||
|
||||
TimedHash *new_items = sys_malloc(new_capacity * sizeof(TimedHash));
|
||||
if (new_items == NULL)
|
||||
return -1;
|
||||
|
||||
if (set->capacity > 0) {
|
||||
memcpy(new_items, set->items, set->count * sizeof(set->items[0]));
|
||||
sys_free(set->items);
|
||||
}
|
||||
|
||||
set->items = new_items;
|
||||
set->capacity = new_capacity;
|
||||
}
|
||||
|
||||
set->items[set->count++] = (TimedHash) { hash, time };
|
||||
return 0;
|
||||
}
|
||||
|
||||
void timed_hash_set_remove(TimedHashSet *set, SHA256 hash)
|
||||
{
|
||||
int idx = timed_hash_set_find(set, hash);
|
||||
if (idx >= 0) {
|
||||
// Remove by shifting remaining items
|
||||
if (idx < set->count - 1) {
|
||||
memmove(&set->items[idx], &set->items[idx + 1],
|
||||
(set->count - idx - 1) * sizeof(set->items[0]));
|
||||
}
|
||||
set->count--;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
#ifndef HASH_SET_INCLUDED
|
||||
#define HASH_SET_INCLUDED
|
||||
|
||||
#include "basic.h"
|
||||
|
||||
typedef struct {
|
||||
SHA256 *items;
|
||||
int count;
|
||||
int capacity;
|
||||
} HashSet;
|
||||
|
||||
typedef struct {
|
||||
SHA256 hash;
|
||||
Time time;
|
||||
} TimedHash;
|
||||
|
||||
typedef struct {
|
||||
TimedHash *items;
|
||||
int count;
|
||||
int capacity;
|
||||
} TimedHashSet;
|
||||
|
||||
void hash_set_init (HashSet *set);
|
||||
void hash_set_free (HashSet *set);
|
||||
void hash_set_clear (HashSet *set);
|
||||
int hash_set_insert (HashSet *set, SHA256 hash);
|
||||
bool hash_set_remove (HashSet *set, SHA256 hash);
|
||||
int hash_set_merge (HashSet *dst, HashSet src);
|
||||
void hash_set_remove_set(HashSet *dst, HashSet src);
|
||||
bool hash_set_contains (HashSet *set, SHA256 hash);
|
||||
|
||||
void timed_hash_set_init (TimedHashSet *set);
|
||||
void timed_hash_set_free (TimedHashSet *set);
|
||||
int timed_hash_set_find (TimedHashSet *set, SHA256 hash);
|
||||
int timed_hash_set_insert (TimedHashSet *set, SHA256 hash, Time time);
|
||||
void timed_hash_set_remove (TimedHashSet *set, SHA256 hash);
|
||||
|
||||
#endif // HASH_SET_INCLUDED
|
||||
+5
-96
@@ -94,13 +94,14 @@ static char *message_type_to_str(uint16_t type)
|
||||
case MESSAGE_TYPE_WRITE_SUCCESS: return "WRITE_SUCCESS";
|
||||
|
||||
// Metadata server -> Chunk server
|
||||
case MESSAGE_TYPE_STATE_UPDATE: return "STATE_UPDATE";
|
||||
case MESSAGE_TYPE_SYNC_2: return "SYNC_2";
|
||||
case MESSAGE_TYPE_SYNC_4: return "SYNC_4";
|
||||
case MESSAGE_TYPE_DOWNLOAD_LOCATIONS: return "DOWNLOAD_LOCATIONS";
|
||||
|
||||
// Chunk server -> Metadata server
|
||||
case MESSAGE_TYPE_AUTH: return "AUTH";
|
||||
case MESSAGE_TYPE_STATE_UPDATE_ERROR: return "UPDATE_ERROR";
|
||||
case MESSAGE_TYPE_STATE_UPDATE_SUCCESS: return "UPDATE_SUCCESS";
|
||||
case MESSAGE_TYPE_SYNC: return "SYNC";
|
||||
case MESSAGE_TYPE_SYNC_3: return "SYNC_3";
|
||||
|
||||
// Chunk server -> Client
|
||||
case MESSAGE_TYPE_CREATE_CHUNK_ERROR: return "CREATE_CHUNK_ERROR";
|
||||
@@ -418,99 +419,7 @@ void message_dump(FILE *stream, ByteView msg)
|
||||
}
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_DOWNLOAD_CHUNK:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
// Metadata server -> Client
|
||||
|
||||
case MESSAGE_TYPE_CREATE_ERROR:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_CREATE_SUCCESS:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_DELETE_ERROR:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_DELETE_SUCCESS:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_LIST_ERROR:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_LIST_SUCCESS:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_READ_ERROR:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_READ_SUCCESS:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_WRITE_ERROR:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_WRITE_SUCCESS:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
// Metadata server -> Chunk server
|
||||
|
||||
case MESSAGE_TYPE_STATE_UPDATE:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_DOWNLOAD_LOCATIONS:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
// Chunk server -> Metadata server
|
||||
|
||||
case MESSAGE_TYPE_AUTH:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_STATE_UPDATE_ERROR:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_STATE_UPDATE_SUCCESS:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
// Chunk server -> Client
|
||||
|
||||
case MESSAGE_TYPE_CREATE_CHUNK_ERROR:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_CREATE_CHUNK_SUCCESS:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_UPLOAD_CHUNK_ERROR:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_UPLOAD_CHUNK_SUCCESS:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
|
||||
case MESSAGE_TYPE_DOWNLOAD_CHUNK_SUCCESS:
|
||||
default:
|
||||
printf(" (TODO)\n");
|
||||
break;
|
||||
}
|
||||
|
||||
@@ -40,6 +40,7 @@ enum {
|
||||
// Chunk server -> Metadata server
|
||||
MESSAGE_TYPE_AUTH,
|
||||
MESSAGE_TYPE_SYNC,
|
||||
MESSAGE_TYPE_SYNC_3,
|
||||
|
||||
// Chunk server -> Client
|
||||
MESSAGE_TYPE_CREATE_CHUNK_ERROR,
|
||||
|
||||
+28
-87
@@ -1,8 +1,3 @@
|
||||
#include "basic.h"
|
||||
#include "byte_queue.h"
|
||||
#include "config.h"
|
||||
#include "file_tree.h"
|
||||
#include "tcp.h"
|
||||
#define _GNU_SOURCE
|
||||
|
||||
#include <string.h>
|
||||
@@ -12,82 +7,36 @@
|
||||
#include "message.h"
|
||||
#include "metadata_server.h"
|
||||
|
||||
static void hash_list_init(HashList *hash_list)
|
||||
{
|
||||
hash_list->count = 0;
|
||||
hash_list->capacity = 0;
|
||||
hash_list->items = NULL;
|
||||
}
|
||||
|
||||
static void hash_list_free(HashList *hash_list)
|
||||
{
|
||||
sys_free(hash_list->items);
|
||||
}
|
||||
|
||||
static int hash_list_insert(HashList *hash_list, SHA256 hash)
|
||||
{
|
||||
// Avoid duplicates
|
||||
for (int i = 0; i < hash_list->count; i++)
|
||||
if (!memcmp(&hash_list->items[i], &hash, sizeof(SHA256)))
|
||||
return 0; // Already present
|
||||
|
||||
if (hash_list->count == hash_list->capacity) {
|
||||
|
||||
int new_capacity;
|
||||
if (hash_list->items == NULL)
|
||||
new_capacity = 16;
|
||||
else
|
||||
new_capacity = 2 * hash_list->capacity;
|
||||
|
||||
SHA256 *new_items = sys_realloc(hash_list->items, new_capacity * sizeof(SHA256));
|
||||
if (new_items == NULL)
|
||||
return -1;
|
||||
|
||||
hash_list->items = new_items;
|
||||
hash_list->capacity = new_capacity;
|
||||
}
|
||||
|
||||
hash_list->items[hash_list->count++] = hash;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static bool hash_list_contains(HashList *hash_list, SHA256 hash)
|
||||
{
|
||||
for (int j = 0; j < hash_list->count; j++)
|
||||
if (!memcmp(&hash, &hash_list->items[j], sizeof(SHA256)))
|
||||
return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
static void chunk_server_peer_init(ChunkServerPeer *chunk_server, Time current_time)
|
||||
{
|
||||
chunk_server->used = true;
|
||||
chunk_server->auth = false;
|
||||
chunk_server->num_addrs = 0;
|
||||
hash_list_init(&chunk_server->ms_old_list);
|
||||
hash_list_init(&chunk_server->ms_add_list);
|
||||
hash_list_init(&chunk_server->ms_rem_list);
|
||||
hash_set_init(&chunk_server->ms_old_list);
|
||||
hash_set_init(&chunk_server->ms_add_list);
|
||||
hash_set_init(&chunk_server->ms_rem_list);
|
||||
chunk_server->last_sync_time = current_time;
|
||||
chunk_server->last_response_time = current_time;
|
||||
}
|
||||
|
||||
static void chunk_server_peer_free(ChunkServerPeer *chunk_server)
|
||||
{
|
||||
hash_list_free(&chunk_server->ms_rem_list);
|
||||
hash_list_free(&chunk_server->ms_add_list);
|
||||
hash_list_free(&chunk_server->ms_old_list);
|
||||
hash_set_free(&chunk_server->ms_rem_list);
|
||||
hash_set_free(&chunk_server->ms_add_list);
|
||||
hash_set_free(&chunk_server->ms_old_list);
|
||||
chunk_server->used = false;
|
||||
}
|
||||
|
||||
static bool chunk_server_peer_contains(ChunkServerPeer *chunk_server, SHA256 hash)
|
||||
{
|
||||
return hash_list_contains(&chunk_server->old_list, hash)
|
||||
|| hash_list_contains(&chunk_server->add_list, hash);
|
||||
return hash_set_contains(&chunk_server->ms_old_list, hash)
|
||||
|| hash_set_contains(&chunk_server->ms_add_list, hash);
|
||||
}
|
||||
|
||||
static bool chunk_server_peer_load(ChunkServerPeer *chunk_server)
|
||||
{
|
||||
return chunk_server->old_list.count + chunk_server->add_list.count;
|
||||
return chunk_server->ms_old_list.count
|
||||
+ chunk_server->ms_add_list.count;
|
||||
}
|
||||
|
||||
// Returns all chunk servers holding the given chunk
|
||||
@@ -677,7 +626,7 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg)
|
||||
int k = find_chunk_server_by_addr(state, results[i].addrs[j]);
|
||||
if (k == -1) return -1;
|
||||
|
||||
if (hash_list_insert(&state->chunk_servers[k].ms_add_list, new_hashes[i]) < 0)
|
||||
if (hash_set_insert(&state->chunk_servers[k].ms_add_list, new_hashes[i]) < 0)
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
@@ -690,7 +639,7 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg)
|
||||
// Add to rem_list for all chunk servers that have this chunk
|
||||
for (int j = 0; j < state->num_chunk_servers; j++) {
|
||||
if (chunk_server_peer_contains(&state->chunk_servers[j], removed_hash)) {
|
||||
if (!hash_list_insert(&state->chunk_servers[j].ms_rem_list, removed_hash))
|
||||
if (!hash_set_insert(&state->chunk_servers[j].ms_rem_list, removed_hash))
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
@@ -803,14 +752,7 @@ static int process_chunk_server_auth(MetadataServer *state,
|
||||
// we accept all connections that provide valid address information.
|
||||
chunk_server->auth = true;
|
||||
|
||||
ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx);
|
||||
assert(output);
|
||||
|
||||
MessageWriter writer;
|
||||
message_writer_init(&writer, output, MESSAGE_TYPE_CHUNK_LIST_REQUEST);
|
||||
if (!message_writer_free(&writer))
|
||||
return -1;
|
||||
|
||||
// No need to respond
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -854,7 +796,7 @@ static int process_chunk_server_sync(MetadataServer *state,
|
||||
assert(num_holders <= MAX_CHUNK_SERVERS);
|
||||
|
||||
if (num_holders <= state->replication_factor) {
|
||||
if (hash_list_insert(&chunk_server->ms_add_list, hash) < 0) {
|
||||
if (hash_set_insert(&chunk_server->ms_add_list, hash) < 0) {
|
||||
assert(0); // TODO
|
||||
}
|
||||
continue;
|
||||
@@ -863,10 +805,9 @@ static int process_chunk_server_sync(MetadataServer *state,
|
||||
// If the chunk is over-replicated, either don't add
|
||||
// it to the ms_add_list or add it to the ms_rem_list
|
||||
// of some other holder.
|
||||
|
||||
if (hash_list_insert(xxx, hash) < 0) {
|
||||
assert(0); // TODO
|
||||
}
|
||||
//
|
||||
// TODO: For now we don't add it to the ms_add_list,
|
||||
// but there may be a better solution.
|
||||
}
|
||||
|
||||
if (binary_read(&reader, NULL, 1)) // TODO: this should probably be an assertion
|
||||
@@ -926,8 +867,8 @@ static int process_chunk_server_sync_3(MetadataServer *state,
|
||||
MessageWriter writer;
|
||||
message_writer_init(&writer, output, MESSAGE_TYPE_SYNC_4);
|
||||
|
||||
HashList tmp_list;
|
||||
hash_list_init(&tmp_list);
|
||||
HashSet tmp_list;
|
||||
hash_set_init(&tmp_list);
|
||||
|
||||
for (uint32_t i = 0; i < count; i++) {
|
||||
|
||||
@@ -935,7 +876,7 @@ static int process_chunk_server_sync_3(MetadataServer *state,
|
||||
if (!binary_read(&reader, &hash, sizeof(hash)))
|
||||
return -1;
|
||||
|
||||
if (hash_list_insert(&tmp_list, hash) < 0) {
|
||||
if (hash_set_insert(&tmp_list, hash) < 0) {
|
||||
assert(0); // TODO
|
||||
}
|
||||
|
||||
@@ -947,23 +888,23 @@ static int process_chunk_server_sync_3(MetadataServer *state,
|
||||
uint32_t tmp = num_holders;
|
||||
message_write(&writer, &tmp, sizeof(tmp));
|
||||
|
||||
for (int i = 0; i < num_holders; i++)
|
||||
message_write_server_addr(&writer, xxx);
|
||||
for (int j = 0; j < num_holders; j++) {
|
||||
int k = holders[j];
|
||||
message_write_server_addr(&writer, &state->chunk_servers[k]);
|
||||
}
|
||||
}
|
||||
|
||||
if (binary_read(&reader, NULL, 1)) // TODO: this should probably be an assertion
|
||||
return -1;
|
||||
|
||||
if (hash_list_merge(&state->ms_old_list, state->ms_add_list) < 0) {
|
||||
if (hash_set_merge(&chunk_server->ms_old_list, chunk_server->ms_add_list) < 0) {
|
||||
assert(0); // TODO
|
||||
}
|
||||
|
||||
if (hash_list_remove_set(&state->ms_old_list, tmp_list) < 0) {
|
||||
assert(0); // TODO
|
||||
}
|
||||
hash_set_remove_set(&chunk_server->ms_old_list, tmp_list);
|
||||
|
||||
hash_list_free(&state->ms_add_list);
|
||||
state->ms_add_list = tmp_list;
|
||||
hash_set_free(&chunk_server->ms_add_list);
|
||||
chunk_server->ms_add_list = tmp_list;
|
||||
|
||||
if (!message_writer_free(&writer))
|
||||
return -1;
|
||||
|
||||
@@ -5,16 +5,11 @@
|
||||
#include "file_tree.h"
|
||||
#include "config.h"
|
||||
#include "basic.h"
|
||||
#include "hash_set.h"
|
||||
|
||||
#define CONNECTION_TAG_CLIENT -2
|
||||
#define CONNECTION_TAG_UNKNOWN -3
|
||||
|
||||
typedef struct {
|
||||
int count;
|
||||
int capacity;
|
||||
SHA256 *items;
|
||||
} HashList;
|
||||
|
||||
typedef struct {
|
||||
|
||||
bool used;
|
||||
@@ -24,13 +19,13 @@ typedef struct {
|
||||
Address addrs[MAX_SERVER_ADDRS];
|
||||
|
||||
// List of chunks that are known to be held by CS
|
||||
HashList ms_old_list;
|
||||
HashSet ms_old_list; // TODO: rename all *_list symbols to *_set
|
||||
|
||||
// List of chunks that should be held by CS
|
||||
HashList ms_add_list;
|
||||
HashSet ms_add_list;
|
||||
|
||||
// List of chunks that may be held by CS but should removed from it
|
||||
HashList ms_rem_list;
|
||||
HashSet ms_rem_list;
|
||||
|
||||
// Time when last STATE_UPDATE was sent
|
||||
Time last_sync_time;
|
||||
|
||||
Reference in New Issue
Block a user