X-Git-Url: http://git.mutantstargoat.com/user/nuclear/?p=assman;a=blobdiff_plain;f=src%2Fmod_url.c;fp=src%2Fmod_url.c;h=5c0f81f1b9755a370f5559c4fc0a01474228cb6f;hp=c8c62cbfbf1e2257526edeaffbb72da650354675;hb=47f73ee9117686e9d3d0b5bf6c664215dfe1bf8d;hpb=326501355589766ea6f9097824b1850d0beae414

diff --git a/src/mod_url.c b/src/mod_url.c
index c8c62cb..5c0f81f 100644
--- a/src/mod_url.c
+++ b/src/mod_url.c
@@ -1,71 +1,329 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <errno.h>
+#include "assman_impl.h"
+#include "tpool.h"
+#include "md4.h"
 
 #ifdef BUILD_MOD_URL
+#include <pthread.h>
 #include <curl/curl.h>
 
+enum {
+	DL_UNKNOWN,
+	DL_STARTED,
+	DL_ERROR,
+	DL_DONE
+};
+
+struct file_info {
+	char *url;
+	char *cache_fname;
+
+	FILE *cache_file;
+
+	/* fopen-thread waits until the state becomes known (request starts transmitting or fails) */
+	int state;
+	pthread_cond_t state_cond;
+	pthread_mutex_t state_mutex;
+};
+
+static void *fop_open(const char *fname, void *udata);
+static void fop_close(void *fp, void *udata);
+static long fop_seek(void *fp, long offs, int whence, void *udata);
+static long fop_read(void *fp, void *buf, long size, void *udata);
+
+static void exit_cleanup(void);
+static void download(void *data);
+static size_t recv_callback(char *ptr, size_t size, size_t nmemb, void *udata);
 static const char *get_temp_dir(void);
 
 static char *tmpdir, *cachedir;
+static struct thread_pool *tpool;
+static CURL **curl;
 
 struct ass_fileops *ass_alloc_url(const char *url)
 {
 	static int done_init;
 	struct ass_fileops *fop;
+	int i;
 
 	if(!done_init) {
 		curl_global_init(CURL_GLOBAL_ALL);
-		atexit(curl_global_cleanup);
-		done_init = 1;
+		atexit(exit_cleanup);
 
 		if(!*ass_mod_url_cachedir) {
 			strcpy(ass_mod_url_cachedir, "assman_cache");
 		}
-		tmpdir = get_temp_dir();
+		tmpdir = (char*)get_temp_dir();
 		if(!(cachedir = malloc(strlen(ass_mod_url_cachedir) + strlen(tmpdir) + 2))) {
 			fprintf(stderr, "assman: failed to allocate cachedir path buffer\n");
-			return 0;
+			goto init_failed;
 		}
 		sprintf(cachedir, "%s/%s", tmpdir, ass_mod_url_cachedir);
+
+		if(ass_mod_url_max_threads <= 0) {
+			ass_mod_url_max_threads = 8;
+		}
+
+		if(!(curl = calloc(ass_mod_url_max_threads, sizeof *curl))) {
+			perror("assman: failed to allocate curl context table");
+			goto init_failed;
+		}
+		for(i=0; i<ass_mod_url_max_threads; i++) {
+			if(!(curl[i] = curl_easy_init())) {
+				goto init_failed;
+			}
+			curl_easy_setopt(curl[i], CURLOPT_WRITEFUNCTION, recv_callback);
+		}
+
+		if(!(tpool = ass_tpool_create(ass_mod_url_max_threads))) {
+			fprintf(stderr, "assman: failed to create thread pool\n");
+			goto init_failed;
+		}
+
+		done_init = 1;
 	}
 
 	if(!(fop = malloc(sizeof *fop))) {
 		return 0;
 	}
+	if(!(fop->udata = malloc(strlen(url) + 1))) {
+		free(fop);
+		return 0;
+	}
+	strcpy(fop->udata, url);
 
-	fop->udata = 0;
 	fop->open = fop_open;
 	fop->close = fop_close;
 	fop->seek = fop_seek;
 	fop->read = fop_read;
 	return fop;
+
+init_failed:
+	free(cachedir);
+	if(curl) {
+		for(i=0; i<ass_mod_url_max_threads; i++) {
+			if(curl[i]) {
+				curl_easy_cleanup(curl[i]);
+			}
+		}
+		free(curl);
+	}
+	return 0;
 }
 
+static void exit_cleanup(void)
+{
+	int i;
+
+	if(tpool) {
+		ass_tpool_destroy(tpool);
+	}
+	if(curl) {
+		for(i=0; i<ass_mod_url_max_threads; i++) {
+			if(curl[i]) {
+				curl_easy_cleanup(curl[i]);
+			}
+		}
+		free(curl);
+	}
+	curl_global_cleanup();
+}
+
+
 void ass_free_url(struct ass_fileops *fop)
 {
 }
 
+static char *cache_filename(const char *fname, const char *url_prefix)
+{
+	MD4_CTX md4ctx;
+	unsigned char sum[16];
+	char sumstr[33];
+	char *resfname;
+	int i;
+	int fname_len = strlen(fname);
+	int prefix_len = strlen(url_prefix);
+	int url_len = fname_len + prefix_len + 1;
+
+	char *url = alloca(url_len + 1);
+	sprintf(url, "%s/%s", url_prefix, fname);
+
+	MD4Init(&md4ctx);
+	MD4Update(&md4ctx, (unsigned char*)url, url_len);
+	MD4Final((unsigned char*)sum, &md4ctx);
+
+	for(i=0; i<16; i++) {
+		sprintf(sumstr + i * 2, "%x", (unsigned int)sum[i]);
+	}
+	sumstr[32] = 0;
+
+	prefix_len = strlen(cachedir);
+	if(!(resfname = malloc(prefix_len + fname_len + 20))) {
+		return 0;
+	}
+	sprintf(resfname, "%s/%s-%s", cachedir, fname, sumstr);
+	return resfname;
+}
+
 static void *fop_open(const char *fname, void *udata)
 {
-	return 0;	/* TODO */
+	struct file_info *file;
+	int state;
+
+	if(!(file = malloc(sizeof *file))) {
+		ass_errno = ENOMEM;
+		return 0;
+	}
+	if(!(file->cache_fname = cache_filename(fname, udata))) {
+		free(file);
+		ass_errno = ENOMEM;
+		return 0;
+	}
+	if(!(file->cache_file = fopen(file->cache_fname, "wb"))) {
+		fprintf(stderr, "assman: mod_url: failed to open cache file (%s) for writing: %s\n",
+				file->cache_fname, strerror(errno));
+		ass_errno = errno;
+		free(file->cache_fname);
+		free(file);
+		return 0;
+	}
+
+	file->state = DL_UNKNOWN;
+	pthread_mutex_init(&file->state_mutex, 0);
+	pthread_cond_init(&file->state_cond, 0);
+
+	ass_tpool_enqueue(tpool, file, download, 0);
+
+	/* wait until the file changes state */
+	pthread_mutex_lock(&file->state_mutex);
+	while(file->state == DL_UNKNOWN) {
+		pthread_cond_wait(&file->state_cond, &file->state_mutex);
+	}
+	state = file->state;
+	pthread_mutex_unlock(&file->state_mutex);
+
+	if(state == DL_ERROR) {
+		/* the worker stopped, so we can safely cleanup and return error */
+		fclose(file->cache_file);
+		remove(file->cache_fname);
+		free(file->cache_fname);
+		pthread_cond_destroy(&file->state_cond);
+		pthread_mutex_destroy(&file->state_mutex);
+		free(file);
+		ass_errno = ENOENT;	/* TODO: differentiate between 403 and 404 */
+		return 0;
+	}
+	return file;
+}
+
+static void wait_done(struct file_info *file)
+{
+	pthread_mutex_lock(&file->state_mutex);
+	while(file->state != DL_DONE && file->state != DL_ERROR) {
+		pthread_cond_wait(&file->state_cond, &file->state_mutex);
+	}
+	pthread_mutex_unlock(&file->state_mutex);
 }
 
 static void fop_close(void *fp, void *udata)
 {
+	struct file_info *file = fp;
+
+	wait_done(file);	/* TODO: stop download instead of waiting to finish */
+
+	fclose(file->cache_file);
+	if(file->state == DL_ERROR) remove(file->cache_fname);
+	free(file->cache_fname);
+	pthread_cond_destroy(&file->state_cond);
+	pthread_mutex_destroy(&file->state_mutex);
+	free(file);
 }
 
 static long fop_seek(void *fp, long offs, int whence, void *udata)
 {
-	return 0;
+	struct file_info *file = fp;
+	wait_done(file);
+
+	fseek(file->cache_file, offs, whence);
+	return ftell(file->cache_file);
 }
 
 static long fop_read(void *fp, void *buf, long size, void *udata)
 {
-	return 0;
+	struct file_info *file = fp;
+	wait_done(file);
+
+	return fread(buf, 1, size, file->cache_file);
 }
 
-#else
+/* this is the function called by the worker threads to perform the download
+ * signal state changes, and prepare the cache file for reading
+ */
+static void download(void *data)
+{
+	int tid, res;
+	struct file_info *file = data;
+
+	tid = ass_tpool_thread_id(tpool);
+
+	curl_easy_setopt(curl[tid], CURLOPT_URL, file->url);
+	curl_easy_setopt(curl[tid], CURLOPT_WRITEDATA, file);
+	res = curl_easy_perform(curl[tid]);
+
+	pthread_mutex_lock(&file->state_mutex);
+	if(res == CURLE_OK) {
+		file->state = DL_DONE;
+		fclose(file->cache_file);
+		if(!(file->cache_file = fopen(file->cache_fname, "rb"))) {
+			fprintf(stderr, "assman: failed to reopen cache file (%s) for reading: %s\n",
+					file->cache_fname, strerror(errno));
+			file->state = DL_ERROR;
+		}
+	} else {
+		file->state = DL_ERROR;
+	}
+	pthread_cond_broadcast(&file->state_cond);
+	pthread_mutex_unlock(&file->state_mutex);
+}
+
+/* this function is called by curl to pass along downloaded data chunks */
+static size_t recv_callback(char *ptr, size_t size, size_t count, void *udata)
+{
+	struct file_info *file = udata;
+
+	pthread_mutex_lock(&file->state_mutex);
+	if(file->state == DL_UNKNOWN) {
+		file->state = DL_STARTED;
+		pthread_cond_broadcast(&file->state_cond);
+	}
+	pthread_mutex_unlock(&file->state_mutex);
+
+	return fwrite(ptr, size, count, file->cache_file);
+}
+
+#ifdef WIN32
+#include <windows.h>
+
+static const char *get_temp_dir(void)
+{
+	static char buf[MAX_PATH + 1];
+	GetTempPathA(MAX_PATH + 1, buf);
+	return buf;
+}
+#else	/* UNIX */
+static const char *get_temp_dir(void)
+{
+	char *env = getenv("TMPDIR");
+	return env ? env : "/tmp";
+}
+#endif
+
+
+
+#else	/* don't build mod_url */
 struct ass_fileops *ass_alloc_url(const char *url)
 {
 	fprintf(stderr, "assman: compiled without URL asset source support\n");