Vine: Send cache-update on all file types. (#4128)

* - Manager now puts replicas of all types into PENDING state when sent.
- Worker now responds with cache-update for all file types.

* format

---------

Co-authored-by: Benjamin Tovar <[email protected]>

Author: dthain

taskvine/src/manager/vine_manager_put.c

@@ -441,24 +441,12 @@ static vine_result_code_t vine_manager_put_input_file_if_needed(struct vine_mana
 	/* Now send the actual file. */
 	vine_result_code_t result = vine_manager_put_input_file(q, w, t, m, file_to_send);
 
-	/* If the send succeeded, then record it in the worker */
-	if (result == VINE_SUCCESS) {
-		struct vine_file_replica *replica = vine_file_replica_table_get_or_create(q, w, f->cached_name, f->type, f->cache_level, f->size, f->mtime);
+	/* If the send succeeded, then note that we have a PENDING replica */
+	/* If will be marked as READY when a cache-update message comes back. */
 
-		switch (file_to_send->type) {
-		case VINE_URL:
-		case VINE_TEMP:
-			/* For these types, a cache-update will arrive when the replica actually exists. */
-			replica->state = VINE_FILE_REPLICA_STATE_CREATING;
-			break;
-		case VINE_FILE:
-		case VINE_MINI_TASK:
-		case VINE_BUFFER:
-			/* For these types, we sent the data, so we know it exists. */
-			replica->state = VINE_FILE_REPLICA_STATE_READY;
-			f->state = VINE_FILE_STATE_CREATED;
-			break;
-		}
+	if (result == VINE_SUCCESS) {
+		struct vine_file_replica *replica = vine_file_replica_create(f->type, f->cache_level, f->size, f->mtime);
+		vine_file_replica_table_insert(q, w, f->cached_name, replica);
 	}
 
 	return result;

taskvine/src/manager/vine_protocol.h

@@ -13,7 +13,7 @@ worker, and catalog, but should not be visible to the public user API.
 #ifndef VINE_PROTOCOL_H
 #define VINE_PROTOCOL_H
 
-#define VINE_PROTOCOL_VERSION 12
+#define VINE_PROTOCOL_VERSION 13
 
 #define VINE_LINE_MAX 4096       /**< Maximum length of a vine message line. */
 

taskvine/src/worker/vine_cache.c

@@ -122,7 +122,7 @@ void vine_cache_scan(struct vine_cache *c, struct link *manager)
 	char *cachename;
 	HASH_TABLE_ITERATE(c->table, cachename, f)
 	{
-		vine_worker_send_cache_update(manager, cachename, f->original_type, f->cache_level, f->size, f->mode, f->transfer_time, f->start_time);
+		vine_worker_send_cache_update(manager, cachename, f);
 	}
 }
 
@@ -262,7 +262,7 @@ and writing out the metadata to the proper location.
 */
 
 int vine_cache_add_file(
-		struct vine_cache *c, const char *cachename, const char *transfer_path, vine_cache_level_t level, int mode, uint64_t size, time_t mtime, timestamp_t transfer_time)
+		struct vine_cache *c, const char *cachename, const char *transfer_path, vine_cache_level_t level, int mode, uint64_t size, time_t mtime, timestamp_t start_time, timestamp_t transfer_time, struct link *manager)
 {
 	char *data_path = vine_cache_data_path(c, cachename);
 	char *meta_path = vine_cache_meta_path(c, cachename);
@@ -285,12 +285,16 @@ int vine_cache_add_file(
 		f->size = size;
 		f->mtime = mtime;
 		f->transfer_time = transfer_time;
+		f->start_time = start_time;
 
 		/* File has data and is ready to use. */
 		f->status = VINE_CACHE_STATUS_READY;
 
 		vine_cache_file_save_metadata(f, meta_path);
 
+		/* Inform the manager that we now have the file */
+		vine_worker_send_cache_update(manager, cachename, f);
+
 		result = 1;
 	} else {
 		result = 0;
@@ -829,7 +833,7 @@ static void vine_cache_check_outputs(struct vine_cache *c, struct vine_cache_fil
 		debug(D_VINE, "cache: measuring %s", transfer_path);
 		if (vine_cache_file_measure_metadata(transfer_path, &mode, &size, &mtime)) {
 			debug(D_VINE, "cache: created %s with size %lld in %lld usec", cachename, (long long)size, (long long)transfer_time);
-			if (vine_cache_add_file(c, cachename, transfer_path, f->cache_level, mode, size, mtime, transfer_time)) {
+			if (vine_cache_add_file(c, cachename, transfer_path, f->cache_level, mode, size, mtime, f->start_time, transfer_time, manager)) {
 				f->status = VINE_CACHE_STATUS_READY;
 			} else {
 				debug(D_VINE, "cache: unable to move %s to %s: %s\n", transfer_path, cache_path, strerror(errno));
@@ -849,7 +853,7 @@ static void vine_cache_check_outputs(struct vine_cache *c, struct vine_cache_fil
 
 	if (manager) {
 		if (f->status == VINE_CACHE_STATUS_READY) {
-			vine_worker_send_cache_update(manager, cachename, f->original_type, f->cache_level, f->size, f->mode, transfer_time, f->start_time);
+			/* a positive cache-update message was sent by vine_cache_add_file */
 		} else {
 			char *error_path = vine_cache_error_path(c, cachename);
 			char *error_message = NULL;

taskvine/src/worker/vine_cache.h

@@ -55,7 +55,7 @@ char *vine_cache_meta_path( struct vine_cache *c, const char *cachename );
 char *vine_cache_transfer_path( struct vine_cache *c, const char *cachename );
 char *vine_cache_error_path( struct vine_cache *c, const char *cachename );
 
-int vine_cache_add_file( struct vine_cache *c, const char *cachename, const char *transfer_path, vine_cache_level_t level, int mode, uint64_t size, time_t mtime, timestamp_t transfer_time );
+int vine_cache_add_file( struct vine_cache *c, const char *cachename, const char *transfer_path, vine_cache_level_t level, int mode, uint64_t size, time_t mtime, timestamp_t start_time, timestamp_t transfer_time, struct link *manager );
 int vine_cache_add_transfer( struct vine_cache *c, const char *cachename, const char *source, vine_cache_level_t level, int mode, uint64_t size, vine_cache_flags_t flags );
 int vine_cache_add_mini_task( struct vine_cache *c, const char *cachename, const char *source, struct vine_task *mini_task, vine_cache_level_t level, int mode, uint64_t size );
 

taskvine/src/worker/vine_sandbox.c

@@ -201,9 +201,8 @@ static int stage_output_file(struct vine_process *p, struct vine_mount *m, struc
 	debug(D_VINE, "output: measuring %s", sandbox_path);
 	if (vine_cache_file_measure_metadata(sandbox_path, &mode, &size, &mtime)) {
 		debug(D_VINE, "output: moving %s to %s", sandbox_path, cache_path);
-		if (vine_cache_add_file(cache, f->cached_name, sandbox_path, f->cache_level, mode, size, mtime, transfer_time)) {
+		if (vine_cache_add_file(cache, f->cached_name, sandbox_path, f->cache_level, mode, size, mtime, p->execution_start, transfer_time, manager)) {
 			f->size = size;
-			vine_worker_send_cache_update(manager, f->cached_name, f->type, f->cache_level, f->size, mode, transfer_time, p->execution_start);
 			result = 1;
 		} else {
 			debug(D_VINE, "output: unable to move %s to %s: %s\n", sandbox_path, cache_path, strerror(errno));

taskvine/src/worker/vine_worker.c

@@ -5,6 +5,7 @@ See the file COPYING for details.
 */
 
 #include "vine_cache.h"
+#include "vine_cache_file.h"
 #include "vine_catalog.h"
 #include "vine_file.h"
 #include "vine_gpus.h"
@@ -495,8 +496,7 @@ Send an asynchronmous message to the manager indicating that an item was success
 its size in bytes and transfer time in usec.
 */
 
-void vine_worker_send_cache_update(struct link *manager, const char *cachename, vine_file_type_t type, vine_cache_level_t cache_level, int64_t size, time_t mtime,
-		timestamp_t transfer_time, timestamp_t transfer_start)
+void vine_worker_send_cache_update(struct link *manager, const char *cachename, struct vine_cache_file *f)
 {
 	char *transfer_id = hash_table_remove(current_transfers, cachename);
 	if (!transfer_id) {
@@ -506,12 +506,12 @@ void vine_worker_send_cache_update(struct link *manager, const char *cachename,
 	send_async_message(manager,
 			"cache-update %s %d %d %lld %lld %lld %lld %s\n",
 			cachename,
-			type,
-			cache_level,
-			(long long)size,
-			(long long)mtime,
-			(long long)transfer_time,
-			(long long)transfer_start,
+			f->original_type,
+			f->cache_level,
+			(long long)f->size,
+			(long long)f->mtime,
+			(long long)f->transfer_time,
+			(long long)f->start_time,
 			transfer_id);
 
 	free(transfer_id);
@@ -986,7 +986,7 @@ static int do_put(struct link *manager, const char *cachename, vine_cache_level_
 	/* XXX actual_size should equal expected size, but only for a simple file, not a dir. */
 
 	if (r) {
-		vine_cache_add_file(cache_manager, cachename, transfer_path, cache_level, mode, actual_size, mtime, stop - start);
+		vine_cache_add_file(cache_manager, cachename, transfer_path, cache_level, mode, actual_size, mtime, start, stop - start, manager);
 	} else {
 		trash_file(transfer_path);
 	}

taskvine/src/worker/vine_worker.h

@@ -5,11 +5,12 @@
 
 #include "vine_workspace.h"
 #include "vine_worker_options.h"
+#include "vine_cache_file.h"
 
 #include "timestamp.h"
 #include "link.h"
 
-void vine_worker_send_cache_update( struct link *manager, const char *cachename, vine_file_type_t type, vine_cache_level_t cache_level, int64_t size, time_t mtime, timestamp_t transfer_time, timestamp_t transfer_start );
+void vine_worker_send_cache_update( struct link *manager, const char *cachename, struct vine_cache_file *f );
 void vine_worker_send_cache_invalid( struct link *manager, const char *cachename, const char *message );
 
 extern struct vine_workspace *workspace;