Progress
This commit is contained in:
@@ -6,3 +6,5 @@
|
||||
- When a client received metadata and starts reading or writing to a chunk server, it should try connecting to all addresses of a chunk servers, not just the first one.
|
||||
- Return the number of bytes read or written in the TinyDFS_Result struct
|
||||
- Make parallel uploads/downloads configurable
|
||||
- Recalculate next write locations whenever a write occurs, not at each read
|
||||
- Make sure there are no mixups with the default tag value for connections, which I think is -1?
|
||||
|
||||
+40
@@ -1,3 +1,5 @@
|
||||
#include <string.h>
|
||||
|
||||
#include "basic.h"
|
||||
#include "system.h"
|
||||
|
||||
@@ -54,3 +56,41 @@ Time get_current_time(void)
|
||||
}
|
||||
#endif
|
||||
}
|
||||
|
||||
bool getargb(int argc, char **argv, char *name)
|
||||
{
|
||||
for (int i = 0; i < argc; i++)
|
||||
if (!strcmp(argv[i], name))
|
||||
return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
string getargs(int argc, char **argv, char *name, char *fallback)
|
||||
{
|
||||
for (int i = 0; i < argc; i++)
|
||||
if (!strcmp(argv[i], name)) {
|
||||
i++;
|
||||
if (i == argc)
|
||||
break;
|
||||
return (string) { argv[i], strlen(argv[i]) };
|
||||
}
|
||||
return (string) { fallback, strlen(fallback) };
|
||||
}
|
||||
|
||||
int getargi(int argc, char **argv, char *name, int fallback)
|
||||
{
|
||||
for (int i = 0; i < argc; i++)
|
||||
if (!strcmp(argv[i], name)) {
|
||||
|
||||
i++;
|
||||
if (i == argc)
|
||||
break;
|
||||
|
||||
int tmp = atoi(argv[i]);
|
||||
if (tmp == 0 && argv[i][0] != '0') // best effort
|
||||
break;
|
||||
|
||||
return tmp;
|
||||
}
|
||||
return fallback;
|
||||
}
|
||||
|
||||
@@ -25,4 +25,8 @@ typedef uint64_t Time;
|
||||
bool streq(string s1, string s2);
|
||||
Time get_current_time(void);
|
||||
|
||||
bool getargb(int argc, char **argv, char *name);
|
||||
string getargs(int argc, char **argv, char *name, char *fallback);
|
||||
int getargi(int argc, char **argv, char *name, int fallback);
|
||||
|
||||
#endif // BASIC_INCLUDED
|
||||
|
||||
+31
-9
@@ -766,15 +766,18 @@ process_client_message(ChunkServer *state, int conn_idx, uint16_t type, ByteView
|
||||
|
||||
int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts, struct pollfd *polled, int *timeout)
|
||||
{
|
||||
(void) argc;
|
||||
(void) argv;
|
||||
string addr = getargs(argc, argv, "--addr", "127.0.0.1");
|
||||
int port = getargi(argc, argv, "--port", 8081);
|
||||
string path = getargs(argc, argv, "--path", "chunk_server_data/");
|
||||
|
||||
char addr[] = "127.0.0.1";
|
||||
uint16_t port = 8080;
|
||||
string path = S("chunk_server_data_0/");
|
||||
string remote_addr = getargs(argc, argv, "--remote-addr", "127.0.0.1");
|
||||
int remote_port = getargi(argc, argv, "--remote-port", 8080);
|
||||
|
||||
char metadata_server_addr[] = "127.0.0.1";
|
||||
uint16_t metadata_server_port = 8081;
|
||||
if (port <= 0 || port >= 1<<16)
|
||||
return -1;
|
||||
|
||||
if (remote_port <= 0 || remote_port >= 1<<16)
|
||||
return -1;
|
||||
|
||||
tcp_context_init(&state->tcp);
|
||||
|
||||
@@ -793,18 +796,37 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts
|
||||
state->downloading = false;
|
||||
pending_download_list_init(&state->pending_download_list);
|
||||
|
||||
char tmp[1<<10];
|
||||
if (remote_addr.len >= (int) sizeof(tmp)) {
|
||||
tcp_context_free(&state->tcp);
|
||||
return -1;
|
||||
}
|
||||
memcpy(tmp, remote_addr.ptr, remote_addr.len);
|
||||
tmp[remote_addr.len] = '\0';
|
||||
|
||||
// Initialize metadata server address
|
||||
// // TODO: This should also support IPv6
|
||||
state->metadata_server_addr.is_ipv4 = true;
|
||||
if (inet_pton(AF_INET, metadata_server_addr, &state->metadata_server_addr.ipv4) != 1) {
|
||||
if (inet_pton(AF_INET, tmp, &state->metadata_server_addr.ipv4) != 1) {
|
||||
tcp_context_free(&state->tcp);
|
||||
chunk_store_free(&state->store);
|
||||
return -1;
|
||||
}
|
||||
state->metadata_server_addr.port = metadata_server_port;
|
||||
state->metadata_server_addr.port = remote_port;
|
||||
|
||||
state->metadata_server_disconnect_time = 0;
|
||||
|
||||
printf("Chunk server set up (local=%.*s:%d, remote=%.*s:%d, path=%.*s)\n",
|
||||
addr.len,
|
||||
addr.ptr,
|
||||
port,
|
||||
remote_addr.len,
|
||||
remote_addr.ptr,
|
||||
remote_port,
|
||||
path.len,
|
||||
path.ptr
|
||||
);
|
||||
|
||||
*timeout = -1; // No timeout needed for chunk server initially
|
||||
return tcp_register_events(&state->tcp, contexts, polled);
|
||||
}
|
||||
|
||||
+393
-8
@@ -1,3 +1,4 @@
|
||||
#include "basic.h"
|
||||
#include <assert.h>
|
||||
#include <string.h>
|
||||
#include <stdlib.h>
|
||||
@@ -38,6 +39,57 @@ typedef enum {
|
||||
OPERATION_TYPE_WRITE,
|
||||
} OperationType;
|
||||
|
||||
typedef struct {
|
||||
SHA256 hash;
|
||||
char * src;
|
||||
int len;
|
||||
int num_holders;
|
||||
Address holders[MAX_CHUNK_HOLDERS][MAX_CHUNK_SERVER_ADDR];
|
||||
} WriteChunk;
|
||||
|
||||
typedef enum {
|
||||
|
||||
// This upload wasn't started yet
|
||||
UPLOAD_WAITING,
|
||||
|
||||
// This upload started
|
||||
UPLOAD_PENDING,
|
||||
|
||||
// This upload was WAITING but then
|
||||
// was marked as IGNORED
|
||||
UPLOAD_IGNORED,
|
||||
|
||||
// Upload was PENDING and FAILED
|
||||
UPLOAD_FAILED,
|
||||
|
||||
// Upload was PENDING, then COMPLETED
|
||||
// successfully
|
||||
UPLOAD_COMPLETED,
|
||||
|
||||
} UploadScheduleStatus;
|
||||
|
||||
typedef struct {
|
||||
UploadScheduleStatus status;
|
||||
|
||||
// Location of the chunk to be patched.
|
||||
// The server local ID is used to indicate
|
||||
// that different addresses refer to the
|
||||
// same server.
|
||||
// The no_hash flag indicates that this is
|
||||
// a new chunk and doesn't need to patch
|
||||
// an old one.
|
||||
int server_lid;
|
||||
bool no_hash;
|
||||
Address address;
|
||||
SHA256 hash;
|
||||
|
||||
// The patch offset and data
|
||||
char *src;
|
||||
int off;
|
||||
int len;
|
||||
|
||||
} UploadSchedule;
|
||||
|
||||
typedef struct {
|
||||
|
||||
OperationType type;
|
||||
@@ -51,6 +103,9 @@ typedef struct {
|
||||
int ranges_count;
|
||||
int num_pending;
|
||||
|
||||
UploadSchedule *uploads;
|
||||
int num_uploads;
|
||||
|
||||
TinyDFS_Result result;
|
||||
} Operation;
|
||||
|
||||
@@ -129,7 +184,6 @@ TinyDFS *tinydfs_init(char *addr, uint16_t port)
|
||||
tdfs->num_chunk_servers = 0;
|
||||
for (int i = 0; i < MAX_CHUNK_SERVERS; i++) {
|
||||
tdfs->chunk_servers[i].used = false;
|
||||
// Note: RequestQueue initialized in get_chunk_server_connection()
|
||||
}
|
||||
|
||||
return tdfs;
|
||||
@@ -885,6 +939,90 @@ static void process_event_for_read(TinyDFS *tdfs,
|
||||
}
|
||||
}
|
||||
|
||||
static int start_upload(Operation *o)
|
||||
{
|
||||
int found = -1;
|
||||
|
||||
// Find a PENDING operation that can be started
|
||||
for (int i = 0; i < o->num_uploads; i++) {
|
||||
|
||||
if (o->status != UPLOAD_PENDING)
|
||||
continue;
|
||||
|
||||
// Can't start uploads of a chunk to the
|
||||
// same server twice.
|
||||
bool invalid = false;
|
||||
for (int j = 0; j < o->num_uploads; j++) {
|
||||
|
||||
if (j == i)
|
||||
continue;
|
||||
|
||||
if (o->uploads[i].server_lid == o->uploads[j].server_lid ||
|
||||
addr_eql(o->uploads[i].address, o->uploads[j].address)) {
|
||||
invalid = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (invalid)
|
||||
continue;
|
||||
|
||||
found = i;
|
||||
}
|
||||
|
||||
if (found < 0)
|
||||
return -1; // No upload can be started at this time
|
||||
|
||||
int ret = get_chunk_server_connection(tdfs, o->uploads[found].address);
|
||||
|
||||
int tag = xxx;
|
||||
|
||||
ByteQueue *output;
|
||||
int ret = tcp_connect(tcp, o->uploads[found].address, tag, &output);
|
||||
if (ret < 0)
|
||||
return -1;
|
||||
|
||||
if (o->uploads[found].no_hash) {
|
||||
|
||||
MessageWriter writer;
|
||||
message_writer_init(&writer, output, MESSAGE_TYPE_CREATE_CHUNK);
|
||||
|
||||
uint32_t chunk_size = xxx;
|
||||
uint32_t target_off = 0;
|
||||
uint32_t target_len = 0;
|
||||
|
||||
message_write(&writer, &chunk_size, sizeof(chunk_size));
|
||||
message_write(&writer, &target_off, sizeof(target_off));
|
||||
message_write(&writer, &target_len, sizeof(target_len));
|
||||
message_write(&writer, data.ptr, data.len);
|
||||
|
||||
if (message_writer_free(&writer) < 0) {
|
||||
// TODO
|
||||
}
|
||||
|
||||
} else {
|
||||
|
||||
MessageWriter writer;
|
||||
message_writer_init(&writer, output, MESSAGE_TYPE_UPLOAD_CHUNK);
|
||||
|
||||
SHA256 target_hash = xxx;
|
||||
uint32_t target_off = 0;
|
||||
uint32_t target_len = 0;
|
||||
|
||||
message_write(&writer, &target_hash, sizeof(target_hash));
|
||||
message_write(&writer, &target_off, sizeof(target_off));
|
||||
message_write(&writer, &target_len, sizeof(target_len));
|
||||
message_write(&writer, data.ptr, data.len);
|
||||
|
||||
if (message_writer_free(&writer) < 0) {
|
||||
// TODO
|
||||
}
|
||||
}
|
||||
|
||||
o->uploads[found].status = UPLOAD_PENDING;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void process_event_for_write(TinyDFS *tdfs,
|
||||
int opidx, int request_tag, ByteView msg)
|
||||
{
|
||||
@@ -896,15 +1034,262 @@ static void process_event_for_write(TinyDFS *tdfs,
|
||||
switch (request_tag) {
|
||||
|
||||
case TAG_RETRIEVE_METADATA_FOR_WRITE:
|
||||
// Process metadata response and initiate chunk uploads
|
||||
// This would involve:
|
||||
// 1. Parsing the metadata response (chunk locations, hashes)
|
||||
// 2. Computing new chunk data by patching existing chunks
|
||||
// 3. Uploading new chunks to chunk servers
|
||||
// 4. Committing the write to the metadata server with new hashes
|
||||
// For now, this operation is not fully implemented
|
||||
{
|
||||
// We are expecting one of:
|
||||
// MESSAGE_TYPE_READ_ERROR
|
||||
// MESSAGE_TYPE_READ_SUCCESS
|
||||
|
||||
BinaryReader reader = { msg.ptr, msg.len, 0 };
|
||||
|
||||
if (!binary_read(&reader, NULL, sizeof(uint16_t))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
uint16_t type;
|
||||
if (!binary_read(&reader, &type, sizeof(type))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
if (type != MESSAGE_TYPE_READ_SUCCESS) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
if (!binary_read(&reader, NULL, sizeof(uint32_t))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
uint32_t chunk_size;
|
||||
if (!binary_read(&reader, &chunk_size, sizeof(chunk_size))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
uint32_t num_hashes;
|
||||
if (!binary_read(&reader, &num_hashes, sizeof(num_hashes))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
// TODO: !!! IMPORTANT !!! This should also account for new chunks, not patched. It does not do so at the moment
|
||||
// TODO: This may overestimate by a lot the actual memory required by the array
|
||||
client->operations[opidx].uploads = sys_malloc(num_hashes * MAX_CHUNK_HOLDERS * MAX_CHUNK_SERVER_ADDR * sizeof(UploadSchedule));
|
||||
if (client->operations[opidx].uploads == NULL) {
|
||||
// TODO
|
||||
}
|
||||
|
||||
int next_server_lid = 0;
|
||||
client->operations[opidx].num_uploads = 0;
|
||||
for (uint32_t i = 0; i < num_hashes; i++) {
|
||||
|
||||
void *src = xxx;
|
||||
int off = xxx;
|
||||
int len = xxx;
|
||||
|
||||
SHA256 hash;
|
||||
if (!binary_read(&reader, &hash, sizeof(hash))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
uint32_t num_holders;
|
||||
if (!binary_read(&reader, &num_holders, sizeof(num_holders))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
for (uint32_t j = 0; j < num_holders; j++) {
|
||||
|
||||
int server_lid = next_server_lid;
|
||||
next_server_lid++;
|
||||
|
||||
uint32_t num_ipv4;
|
||||
if (!binary_read(&reader, &num_ipv4, sizeof(num_ipv4))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
for (uint32_t k = 0; k < num_ipv4; k++) {
|
||||
|
||||
IPv4 ipv4;
|
||||
if (!binary_read(&reader, &ipv4, sizeof(ipv4))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
uint16_t port;
|
||||
if (!binary_read(&reader, &port, sizeof(port))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
UploadSchedule upload;
|
||||
upload.status = UPLOAD_WAITING;
|
||||
upload.server_lid = server_lid;
|
||||
upload.no_hash = false;
|
||||
upload.address.is_ipv4 = true;
|
||||
upload.address.ipv4 = ipv4;
|
||||
upload.address.port = port;
|
||||
upload.hash = hash;
|
||||
upload.src = src;
|
||||
upload.off = off;
|
||||
upload.len = len;
|
||||
int n = client->operations[opidx].num_uploads++;
|
||||
client->operations[opidx].uploads[n] = upload;
|
||||
}
|
||||
|
||||
uint32_t num_ipv6;
|
||||
if (!binary_read(&reader, &num_ipv6, sizeof(num_ipv6))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
for (uint32_t k = 0; k < num_ipv6; k++) {
|
||||
|
||||
IPv6 ipv6;
|
||||
if (!binary_read(&reader, &ipv6, sizeof(ipv6))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
uint16_t port;
|
||||
if (!binary_read(&reader, &port, sizeof(port))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
UploadSchedule upload;
|
||||
upload.status = UPLOAD_WAITING;
|
||||
upload.server_lid = server_lid;
|
||||
upload.no_hash = false;
|
||||
upload.address.is_ipv4 = false;
|
||||
upload.address.ipv6 = ipv6;
|
||||
upload.address.port = port;
|
||||
upload.hash = hash;
|
||||
upload.src = src;
|
||||
upload.off = off;
|
||||
upload.len = len;
|
||||
int n = client->operations[opidx].num_uploads++;
|
||||
client->operations[opidx].uploads[n] = upload;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
uint32_t num_locations;
|
||||
if (!binary_read(&reader, &num_locations, sizeof(num_locations))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
for (uint32_t i = 0; i < num_locations; i++) {
|
||||
|
||||
uint32_t num_ipv4;
|
||||
if (!binary_read(&reader, &num_ipv4, sizeof(num_ipv4))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
for (uint32_t k = 0; k < num_ipv4; k++) {
|
||||
|
||||
IPv4 ipv4;
|
||||
if (!binary_read(&reader, &ipv4, sizeof(ipv4))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
uint16_t port;
|
||||
if (!binary_read(&reader, &port, sizeof(port))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
// TODO
|
||||
}
|
||||
|
||||
uint32_t num_ipv6;
|
||||
if (!binary_read(&reader, &num_ipv6, sizeof(num_ipv6))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
for (uint32_t k = 0; k < num_ipv6; k++) {
|
||||
|
||||
IPv6 ipv6;
|
||||
if (!binary_read(&reader, &ipv6, sizeof(ipv6))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
uint16_t port;
|
||||
if (!binary_read(&reader, &port, sizeof(port))) {
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
// TODO
|
||||
}
|
||||
}
|
||||
|
||||
// Now start the first batch of uploads
|
||||
int started = 0;
|
||||
for (int i = 0; i < xxx; i++) {
|
||||
if (start_upload(&tdfs->operations[opidx]) == 0)
|
||||
started++;
|
||||
}
|
||||
|
||||
if (started == 0) {
|
||||
// We already failed
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
return;
|
||||
}
|
||||
|
||||
// TODO: Now we need to upload the patches to N of the
|
||||
// chunk servers that are holding each old chunk
|
||||
// All new chunks need to be written to the specified
|
||||
// locations at least N times. If any upload fails,
|
||||
// the write fails. If all writes succede, the client
|
||||
// sends the metadata server a WRITE operation
|
||||
// swapping the old hashes with the new ones.
|
||||
//
|
||||
// The algorithm should go like this:
|
||||
// - Iterate over each chunk
|
||||
// - Pick the first N holders of the chunk. If less than N
|
||||
// are available, pick M.
|
||||
// - For each pick, take the first address and start the
|
||||
// chunk upload
|
||||
//
|
||||
// If an upload fails,
|
||||
//
|
||||
//
|
||||
//
|
||||
// example upload schedule:
|
||||
// chunk_A server_A addr_0
|
||||
// chunk_A server_A addr_1
|
||||
// chunk_A server_B addr_0
|
||||
// chunk_A server_B addr_1
|
||||
// chunk_A server_B addr_2
|
||||
// chunk_A server_C addr_0
|
||||
// chunk_B server_D addr_0
|
||||
// chunk_B server_E addr_0
|
||||
// chunk_B server_E addr_1
|
||||
// chunk_B server_F addr_0
|
||||
//
|
||||
// If an upload succedes, all uploads of the chunk to the same server
|
||||
// are removed and if this was the N-th successful upload of a chunk,
|
||||
// all uploads of the same chunk are removed.
|
||||
//
|
||||
// Uploads to the same chunk server with different addresses can't
|
||||
// be parallelized, so
|
||||
|
||||
// The client should not try any random N chunk servers
|
||||
// for upload. It must try all chunk servers until N respond
|
||||
|
||||
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
|
||||
}
|
||||
return;
|
||||
|
||||
default:
|
||||
break;
|
||||
|
||||
+1
-9
@@ -53,17 +53,9 @@ int chunk_server_main(int argc, char **argv)
|
||||
return 0;
|
||||
}
|
||||
|
||||
bool is_leader(int argc, char **argv)
|
||||
{
|
||||
for (int i = 1; i < argc; i++)
|
||||
if (!strcmp(argv[i], "--leader") || !strcmp(argv[i], "-l"))
|
||||
return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
int main(int argc, char **argv)
|
||||
{
|
||||
if (is_leader(argc, argv))
|
||||
if (getargb(argc, argv, "--leader"))
|
||||
return metadata_server_main(argc, argv);
|
||||
else
|
||||
return chunk_server_main(argc, argv);
|
||||
|
||||
+13
-30
@@ -18,45 +18,28 @@ int main(int argc, char **argv)
|
||||
startup_simulation();
|
||||
|
||||
// Spawn metadata server (leader)
|
||||
spawn_simulated_process("--addr 127.0.0.1 8080 --leader");
|
||||
spawn_simulated_process("--addr 127.0.0.1 --port 8080 --leader");
|
||||
|
||||
// Spawn chunk servers
|
||||
spawn_simulated_process("--addr 127.0.0.1 8081");
|
||||
spawn_simulated_process("--addr 127.0.0.1 8082");
|
||||
spawn_simulated_process("--addr 127.0.0.1 8083");
|
||||
spawn_simulated_process("--addr 127.0.0.1 8084");
|
||||
spawn_simulated_process("--addr 127.0.0.1 8085");
|
||||
spawn_simulated_process("--addr 127.0.0.1 8086");
|
||||
spawn_simulated_process("--addr 127.0.0.1 8087");
|
||||
spawn_simulated_process("--addr 127.0.0.1 8088");
|
||||
spawn_simulated_process("--addr 127.0.0.1 8089");
|
||||
spawn_simulated_process("--addr 127.0.0.1 8090");
|
||||
spawn_simulated_process("--addr 127.0.0.1 --port 8081 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_0/");
|
||||
spawn_simulated_process("--addr 127.0.0.1 --port 8082 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_1/");
|
||||
spawn_simulated_process("--addr 127.0.0.1 --port 8083 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_2/");
|
||||
spawn_simulated_process("--addr 127.0.0.1 --port 8084 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_3/");
|
||||
spawn_simulated_process("--addr 127.0.0.1 --port 8085 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_4/");
|
||||
spawn_simulated_process("--addr 127.0.0.1 --port 8086 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_5/");
|
||||
spawn_simulated_process("--addr 127.0.0.1 --port 8087 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_6/");
|
||||
spawn_simulated_process("--addr 127.0.0.1 --port 8088 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_7/");
|
||||
spawn_simulated_process("--addr 127.0.0.1 --port 8089 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_8/");
|
||||
spawn_simulated_process("--addr 127.0.0.1 --port 8090 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_9/");
|
||||
|
||||
// Spawn simulation client
|
||||
spawn_simulated_process("--client --server 127.0.0.1 8080");
|
||||
spawn_simulated_process("--client --remote-addr 127.0.0.1 --remote-port 8080");
|
||||
|
||||
printf("Running simulation (press Ctrl+C to stop)...\n");
|
||||
|
||||
// Run for a limited number of iterations for testing
|
||||
int iteration = 0;
|
||||
int max_iterations = 100000; // Increased to allow client operations to complete
|
||||
while (!simulation_should_stop && iteration < max_iterations) {
|
||||
while (!simulation_should_stop)
|
||||
update_simulation();
|
||||
iteration++;
|
||||
|
||||
// Print progress every 10000 iterations
|
||||
if (iteration % 10000 == 0) {
|
||||
fprintf(stderr, "Iteration %d...\n", iteration);
|
||||
fflush(stderr);
|
||||
}
|
||||
}
|
||||
|
||||
if (iteration >= max_iterations) {
|
||||
printf("\nSimulation stopped after %d iterations\n", max_iterations);
|
||||
}
|
||||
|
||||
cleanup_simulation();
|
||||
printf("Simulation complete!\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
||||
+13
-4
@@ -480,6 +480,9 @@ process_client_read(MetadataServer *state, int conn_idx, ByteView msg)
|
||||
int locations[MAX_CHUNK_SERVERS];
|
||||
int num_locations = choose_servers_for_write(state, locations, state->replication_factor);
|
||||
|
||||
uint32_t tmp = num_locations;
|
||||
message_write(&writer, &tmp, sizeof(tmp));
|
||||
|
||||
for (int j = 0; j < num_locations; j++)
|
||||
message_write_server_addr(&writer, &state->chunk_servers[locations[j]]);
|
||||
|
||||
@@ -752,11 +755,11 @@ static bool is_chunk_server_message_type(uint16_t type)
|
||||
|
||||
int metadata_server_init(MetadataServer *state, int argc, char **argv, void **contexts, struct pollfd *polled, int *timeout)
|
||||
{
|
||||
(void) argc;
|
||||
(void) argv;
|
||||
string addr = getargs(argc, argv, "--addr", "127.0.0.1");
|
||||
int port = getargi(argc, argv, "--port", 8080);
|
||||
|
||||
char addr[] = "127.0.0.1";
|
||||
uint16_t port = 8080;
|
||||
if (port <= 0 || port >= 1<<16)
|
||||
return -1;
|
||||
|
||||
state->replication_factor = 3;
|
||||
if (state->replication_factor > MAX_CHUNK_SERVERS)
|
||||
@@ -778,6 +781,12 @@ int metadata_server_init(MetadataServer *state, int argc, char **argv, void **co
|
||||
return -1;
|
||||
}
|
||||
|
||||
printf("Metadata server set up (local=%.*s:%d)\n",
|
||||
addr.len,
|
||||
addr.ptr,
|
||||
port
|
||||
);
|
||||
|
||||
*timeout = -1; // No timeout needed for metadata server
|
||||
return tcp_register_events(&state->tcp, contexts, polled);
|
||||
}
|
||||
|
||||
@@ -34,11 +34,8 @@ int simulation_client_init(SimulationClient *client, int argc, char **argv,
|
||||
uint16_t port;
|
||||
parse_server_addr(argc, argv, &addr, &port);
|
||||
|
||||
printf("[Client] Initializing TinyDFS client, connecting to %s:%u\n", addr, port);
|
||||
|
||||
client->tdfs = tinydfs_init(addr, port);
|
||||
if (client->tdfs == NULL) {
|
||||
fprintf(stderr, "[Client] Failed to initialize TinyDFS client\n");
|
||||
return -1;
|
||||
}
|
||||
|
||||
@@ -51,7 +48,7 @@ int simulation_client_init(SimulationClient *client, int argc, char **argv,
|
||||
client->list_op = -1;
|
||||
client->delete_op = -1;
|
||||
|
||||
printf("[Client] Initialized successfully\n");
|
||||
printf("Client set up (remote=%s:%d)\n", addr, port);
|
||||
|
||||
*timeout = 0; // Wake up immediately to start processing
|
||||
return tinydfs_process_events(client->tdfs, contexts, polled, 0);
|
||||
|
||||
+2
-18
@@ -241,22 +241,6 @@ static void process_poll_array(Process *process,
|
||||
}
|
||||
}
|
||||
|
||||
static bool is_leader(int argc, char **argv)
|
||||
{
|
||||
for (int i = 0; i < argc; i++)
|
||||
if (!strcmp("--leader", argv[i]) || !strcmp("-l", argv[i]))
|
||||
return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
static bool is_client(int argc, char **argv)
|
||||
{
|
||||
for (int i = 0; i < argc; i++)
|
||||
if (!strcmp("--client", argv[i]) || !strcmp("-c", argv[i]))
|
||||
return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
#define MAX_ARGS 128
|
||||
|
||||
static bool is_space(char c)
|
||||
@@ -303,8 +287,8 @@ int spawn_simulated_process(char *args)
|
||||
cur++;
|
||||
}
|
||||
|
||||
bool leader = is_leader(argc, argv);
|
||||
bool client = is_client(argc, argv);
|
||||
bool leader = getargb(argc, argv, "--leader");
|
||||
bool client = getargb(argc, argv, "--client");
|
||||
|
||||
Process *process = malloc(sizeof(Process));
|
||||
if (process == NULL)
|
||||
|
||||
@@ -24,16 +24,22 @@ bool addr_eql(Address a, Address b)
|
||||
return true;
|
||||
}
|
||||
|
||||
static SOCKET create_listen_socket(char *addr, uint16_t port)
|
||||
static SOCKET create_listen_socket(string addr, uint16_t port)
|
||||
{
|
||||
SOCKET fd = sys_socket(AF_INET, SOCK_STREAM, 0);
|
||||
if (fd == INVALID_SOCKET)
|
||||
return INVALID_SOCKET;
|
||||
|
||||
char tmp[1<<10];
|
||||
if (addr.len >= (int) sizeof(tmp))
|
||||
return INVALID_SOCKET;
|
||||
memcpy(tmp, addr.ptr, addr.len);
|
||||
tmp[addr.len] = '\0';
|
||||
|
||||
struct sockaddr_in bind_buf;
|
||||
bind_buf.sin_family = AF_INET;
|
||||
bind_buf.sin_port = htons(port);
|
||||
if (inet_pton(AF_INET, addr, &bind_buf.sin_addr) != 1)
|
||||
if (inet_pton(AF_INET, tmp, &bind_buf.sin_addr) != 1)
|
||||
return INVALID_SOCKET;
|
||||
|
||||
if (sys_bind(fd, (struct sockaddr*) &bind_buf, sizeof(bind_buf)))
|
||||
@@ -110,7 +116,7 @@ int tcp_index_from_tag(TCP *tcp, int tag)
|
||||
return -1;
|
||||
}
|
||||
|
||||
int tcp_listen(TCP *tcp, char *addr, uint16_t port)
|
||||
int tcp_listen(TCP *tcp, string addr, uint16_t port)
|
||||
{
|
||||
SOCKET listen_fd = create_listen_socket(addr, port);
|
||||
if (listen_fd == INVALID_SOCKET)
|
||||
|
||||
@@ -62,7 +62,7 @@ bool addr_eql(Address a, Address b);
|
||||
void tcp_context_init(TCP *tcp);
|
||||
void tcp_context_free(TCP *tcp);
|
||||
int tcp_index_from_tag(TCP *tcp, int tag);
|
||||
int tcp_listen(TCP *tcp, char *addr, uint16_t port);
|
||||
int tcp_listen(TCP *tcp, string addr, uint16_t port);
|
||||
int tcp_next_message(TCP *tcp, int conn_idx, ByteView *msg, uint16_t *type);
|
||||
void tcp_consume_message(TCP *tcp, int conn_idx);
|
||||
int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd *polled, int num_polled);
|
||||
|
||||
Reference in New Issue
Block a user