From 94d960655e42fa43cae5244ac85ef592a52da051 Mon Sep 17 00:00:00 2001 From: Levi Neely <141506390+lneely@users.noreply.github.com> Date: Sat, 1 Mar 2025 11:07:06 +0100 Subject: [PATCH] ptask namespace and cleanup --- pclsync/pasyncnet.h | 81 --- pclsync/pdiff.c | 26 +- pclsync/pdownload.c | 12 +- pclsync/plocalscan.c | 16 +- pclsync/ppathstatus.c | 2 +- pclsync/pstatus.c | 2 +- pclsync/psyncer.c | 6 +- pclsync/psynclib.c | 25 +- pclsync/{pasyncnet.c => ptask.c} | 971 +++++++++++++++++++------------ pclsync/{ptasks.h => ptask.h} | 121 ++-- pclsync/ptasks.c | 271 --------- pclsync/pupload.c | 2 +- 12 files changed, 690 insertions(+), 845 deletions(-) delete mode 100644 pclsync/pasyncnet.h rename pclsync/{pasyncnet.c => ptask.c} (53%) rename pclsync/{ptasks.h => ptask.h} (55%) delete mode 100644 pclsync/ptasks.c diff --git a/pclsync/pasyncnet.h b/pclsync/pasyncnet.h deleted file mode 100644 index d6d28dc..0000000 --- a/pclsync/pasyncnet.h +++ /dev/null @@ -1,81 +0,0 @@ -/* - Copyright (c) 2015 Anton Titov. - - Copyright (c) 2015 pCloud Ltd. All rights reserved. - - Redistribution and use in source and binary forms, with or without - modification, are permitted provided that the following conditions - are met: Redistributions of source code must retain the above - copyright notice, this list of conditions and the following - disclaimer. Redistributions in binary form must reproduce the above - copyright notice, this list of conditions and the following - disclaimer in the documentation and/or other materials provided with - the distribution. Neither the name of pCloud Ltd nor the names of - its contributors may be used to endorse or promote products derived - from this software without specific prior written permission. - - THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR - A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL pCloud Ltd BE - LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR - BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, - WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE - OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, - EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. -*/ - -#ifndef _PSYNC_ASYNCNET_H -#define _PSYNC_ASYNCNET_H - -#include "psynclib.h" - -#define PSYNC_ASYNC_ERR_FLAG_PERM \ - 0x01 // the error is permanent(ish) and there is no reason to retry -#define PSYNC_ASYNC_ERR_FLAG_RETRY_AS_IS \ - 0x02 // same request may succeed in the future if retried as is -#define PSYNC_ASYNC_ERR_FLAG_SUCCESS \ - 0x04 // like no action performed because of no need - file already exists and - // so on - -#define PSYNC_ASYNC_ERROR_NET 1 -#define PSYNC_ASYNC_ERROR_FILE 2 -#define PSYNC_ASYNC_ERROR_DISK_FULL 3 -#define PSYNC_ASYNC_ERROR_IO 4 -#define PSYNC_ASYNC_ERROR_CHECKSUM 5 - -#define PSYNC_SERVER_ERROR_TOO_BIG 102 -#define PSYNC_SERVER_ERROR_NOT_MOD 104 - -typedef struct { - uint64_t size; - uint64_t hash; - unsigned char sha1hex[40]; -} psync_async_file_result_t; - -typedef struct { - uint32_t error; - uint32_t errorflags; - union { - psync_async_file_result_t file; - }; -} psync_async_result_t; - -typedef void (*psync_async_callback_t)(void *, psync_async_result_t *); - -/* Important! The interface typically expect all passed pointers to be alive - * until the completion callback is called. - */ - -void psync_async_stop(); -int psync_async_download_file(psync_fileid_t fileid, const char *localpath, - psync_async_callback_t cb, void *cbext); -int psync_async_download_file_if_changed(psync_fileid_t fileid, - const char *localpath, uint64_t size, - const void *sha1hex, - psync_async_callback_t cb, - void *cbext); - -#endif diff --git a/pclsync/pdiff.c b/pclsync/pdiff.c index e518d3a..430ebd0 100644 --- a/pclsync/pdiff.c +++ b/pclsync/pdiff.c @@ -63,7 +63,7 @@ #include "psynclib.h" #include "psyncer.h" #include "psys.h" -#include "ptasks.h" +#include "ptask.h" #include "ptimer.h" #include "ptools.h" #include "publiclinks.h" @@ -949,7 +949,7 @@ static void process_createfolder(const binresult *entry) { psync_sql_bind_uint(stmt, 4, row[2]); psync_sql_run(stmt); if (psync_sql_affected_rows() == 1) { - psync_task_create_local_folder(syncid, folderid, localfolderid); + ptask_ldir_mk(syncid, folderid, localfolderid); needdownload = 1; } else { stmt2 = psync_sql_prep_statement("UPDATE syncedfolder SET folderid=? " @@ -1156,7 +1156,7 @@ static void process_modifyfolder(const binresult *entry) { } localfolderid = row[0]; psync_sql_free_result(res); - psync_task_rename_local_folder( + ptask_ldir_rename( psync_get_result_cell(fres2, i, 0), folderid, localfolderid, psync_get_result_cell(fres2, i, 1), name->str); needdownload = 1; @@ -1188,7 +1188,7 @@ static void process_modifyfolder(const binresult *entry) { localfolderid = row[0]; psync_sql_free_result(res); del_synced_folder_rec(folderid, syncid); - psync_task_delete_local_folder_recursive(syncid, folderid, localfolderid); + ptask_ldir_rm_r(syncid, folderid, localfolderid); needdownload = 1; } for (/*i is already=cnt*/; i < fres2->rows; i++) { @@ -1199,7 +1199,7 @@ static void process_modifyfolder(const binresult *entry) { (unsigned long)folderid, (unsigned long)parentfolderid); localfolderid = psync_create_local_folder_in_db( syncid, folderid, psync_get_result_cell(fres2, i, 1), name->str); - psync_task_create_local_folder(syncid, folderid, localfolderid); + ptask_ldir_mk(syncid, folderid, localfolderid); psync_add_folder_for_downloadsync( syncid, psync_get_result_cell(fres2, i, 2), folderid, localfolderid); needdownload = 1; @@ -1253,7 +1253,7 @@ static void process_deletefolder(const binresult *entry) { psync_sql_run_free(stmt); if (psync_sql_affected_rows() == 1) { path = psync_get_path_by_folderid(folderid, NULL); - psync_task_delete_local_folder(row[0], folderid, row[1], path); + ptask_ldir_rm(row[0], folderid, row[1], path); psync_free(path); needdownload = 1; } @@ -1416,7 +1416,7 @@ static void process_createfile(const binresult *entry) { if (!hasit) { debug(D_NOTICE, "downloading file %s with hash %ld to local folder %lu", name->str, (long)hash, (unsigned long)row[1]); - psync_task_download_file_silent(row[0], fileid, row[1], name->str); + ptask_download_q(row[0], fileid, row[1], name->str); needdownload = 1; } else debug(D_NOTICE, @@ -1520,7 +1520,7 @@ static void process_modifyfile(const binresult *entry) { char *path; psync_delete_download_tasks_for_file(fileid, 0, 1); path = psync_get_path_by_fileid(fileid, NULL); - psync_task_delete_local_file(fileid, path); + ptask_lfile_rm(fileid, path); psync_free(path); needdownload = 1; return; @@ -1549,7 +1549,7 @@ static void process_modifyfile(const binresult *entry) { // fres2->rows, oldparentfolderid, parentfolderid); for (i = 0; i < cnt; i++) { if (needrename) { - psync_task_rename_local_file(psync_get_result_cell(fres1, i, 0), + ptask_lfile_rename(psync_get_result_cell(fres1, i, 0), psync_get_result_cell(fres2, i, 0), fileid, psync_get_result_cell(fres1, i, 1), psync_get_result_cell(fres2, i, 1), @@ -1571,7 +1571,7 @@ static void process_modifyfile(const binresult *entry) { "ignoring update for file %s, has correct hash in the database", name->str); else { - psync_task_download_file_silent( + ptask_download_q( psync_get_result_cell(fres2, i, 0), fileid, psync_get_result_cell(fres2, i, 1), name->str); needdownload = 1; @@ -1579,14 +1579,14 @@ static void process_modifyfile(const binresult *entry) { } } for (/*i is already=cnt*/; i < fres2->rows; i++) { - psync_task_download_file_silent( + ptask_download_q( psync_get_result_cell(fres2, i, 0), fileid, psync_get_result_cell(fres2, i, 1), name->str); needdownload = 1; } for (/*i is already=cnt*/; i < fres1->rows; i++) { char *path = psync_get_path_by_fileid(fileid, NULL); - psync_task_delete_local_file_syncid(psync_get_result_cell(fres1, i, 0), + ptask_lfile_rm_id(psync_get_result_cell(fres1, i, 0), fileid, path); psync_delete_download_tasks_for_file( fileid, psync_get_result_cell(fres1, i, 0), 1); @@ -1622,7 +1622,7 @@ static void process_deletefile(const binresult *entry) { psync_delete_download_tasks_for_file(fileid, 0, 1); path = psync_get_path_by_fileid(fileid, NULL); if (likely(path)) { - psync_task_delete_local_file(fileid, path); + ptask_lfile_rm(fileid, path); psync_free(path); needdownload = 1; } diff --git a/pclsync/pdownload.c b/pclsync/pdownload.c index 3debc98..fad069b 100644 --- a/pclsync/pdownload.c +++ b/pclsync/pdownload.c @@ -39,9 +39,7 @@ #include #include - #include "papi.h" -#include "pasyncnet.h" #include "pcallbacks.h" #include "pdownload.h" #include "pfolder.h" @@ -56,7 +54,7 @@ #include "psys.h" #include "pstatus.h" #include "psyncer.h" -#include "ptasks.h" +#include "ptask.h" #include "ptimer.h" #include "pupload.h" #include "putil.h" @@ -973,7 +971,7 @@ static int task_rename_file(psync_syncid_t oldsyncid, psync_syncid_t newsyncid, } psync_sql_free_result(res); if (unlikely_log(!lfileid)) { - psync_task_download_file(newsyncid, fileid, newlocalfolderid, newname); + ptask_download(newsyncid, fileid, newlocalfolderid, newname); return 0; } newfolder = @@ -993,7 +991,7 @@ static int task_rename_file(psync_syncid_t oldsyncid, psync_syncid_t newsyncid, if (errno == ENOENT) { debug(D_WARNING, "renamed from %s to %s failed, downloading", oldpath, newpath); - psync_task_download_file(newsyncid, fileid, newlocalfolderid, newname); + ptask_download(newsyncid, fileid, newlocalfolderid, newname); } else ret = -1; } else { @@ -1315,11 +1313,11 @@ static int task_run_download_file(uint64_t taskid, psync_syncid_t syncid, set_task_inprogress(taskid, 1); if (size <= PSYNC_MAX_SIZE_FOR_ASYNC_DOWNLOAD) { if (dt->localexists) - ret = psync_async_download_file_if_changed( + ret = ptask_download_needed_async( fileid, dt->tmpname, dt->localsize, dt->checksum, finish_async_download_existing, dt); else - ret = psync_async_download_file(fileid, dt->tmpname, + ret = ptask_download_async(fileid, dt->tmpname, finish_async_download, dt); if (ret) { debug(D_WARNING, "async download start failed for %s", dt->localname); diff --git a/pclsync/plocalscan.c b/pclsync/plocalscan.c index f4f87a9..658737b 100644 --- a/pclsync/plocalscan.c +++ b/pclsync/plocalscan.c @@ -51,7 +51,7 @@ #include "pssl.h" #include "pstatus.h" #include "psys.h" -#include "ptasks.h" +#include "ptask.h" #include "ptimer.h" #include "pupload.h" #include "ppath.h" @@ -657,7 +657,7 @@ static void scan_rename_file(sync_folderlist *rnfr, sync_folderlist *rnto) { psync_sql_bind_string(res, 3, rnto->name); psync_sql_bind_uint(res, 4, rnfr->localid); psync_sql_run_free(res); - psync_task_rename_remote_file(rnfr->syncid, rnto->syncid, rnfr->localid, + ptask_rfile_rename(rnfr->syncid, rnto->syncid, rnfr->localid, rnto->localparentfolderid, rnto->name); if (filetoupload) { psync_path_status_sync_folder_task_added_locked(rnto->syncid, @@ -690,7 +690,7 @@ static void scan_upload_file(sync_folderlist *fl) { if (unlikely_log(!psync_sql_affected_rows())) return; localfileid = psync_sql_insertid(); - psync_task_upload_file_silent(fl->syncid, localfileid, fl->name); + ptask_upload_q(fl->syncid, localfileid, fl->name); psync_path_status_sync_folder_task_added(fl->syncid, fl->localparentfolderid); } @@ -707,7 +707,7 @@ static void scan_upload_modified_file(sync_folderlist *fl) { psync_sql_bind_uint(res, 4, fl->mtimenat); psync_sql_bind_uint(res, 5, fl->localid); psync_sql_run_free(res); - psync_task_upload_file_silent(fl->syncid, fl->localid, fl->name); + ptask_upload_q(fl->syncid, fl->localid, fl->name); psync_path_status_sync_folder_task_added(fl->syncid, fl->localparentfolderid); } @@ -737,7 +737,7 @@ static void scan_delete_file(sync_folderlist *fl) { psync_sql_bind_uint(res, 1, fl->localid); psync_sql_run_free(res); if (fileid) - psync_task_delete_remote_file(fl->syncid, fileid); + ptask_rfile_rm(fl->syncid, fileid); psync_path_status_sync_folder_task_completed(syncid, localparentfolderid); } @@ -796,7 +796,7 @@ static void scan_create_folder(sync_folderlist *fl) { psync_sql_run_free(res); if (unlikely_log(!psync_sql_affected_rows())) return; - psync_task_create_remote_folder(fl->syncid, localfolderid, fl->name); + ptask_rdir_mk(fl->syncid, localfolderid, fl->name); return; hasfolder: return; @@ -879,7 +879,7 @@ static void scan_rename_folder(sync_folderlist *rnfr, sync_folderlist *rnto) { (unsigned)rnfr->syncid, (unsigned)rnto->syncid); update_syncid_rec(rnfr->localid, rnto->syncid); } - psync_task_rename_remote_folder(rnfr->syncid, rnto->syncid, rnfr->localid, + ptask_rdir_rename(rnfr->syncid, rnto->syncid, rnfr->localid, rnto->localparentfolderid, rnto->name); } @@ -952,7 +952,7 @@ retry: } delete_local_folder_rec(fl->localid); if (folderid) - psync_task_delete_remote_folder(fl->syncid, folderid); + ptask_rdir_rm(fl->syncid, folderid); } #define check_for_query_cnt() \ diff --git a/pclsync/ppathstatus.c b/pclsync/ppathstatus.c index a9ec005..aea29b3 100644 --- a/pclsync/ppathstatus.c +++ b/pclsync/ppathstatus.c @@ -43,7 +43,7 @@ #include "plist.h" #include "ppathstatus.h" #include "pstatus.h" -#include "ptasks.h" +#include "ptask.h" #include "ptree.h" #include #include diff --git a/pclsync/pstatus.c b/pclsync/pstatus.c index b723519..0d296fd 100644 --- a/pclsync/pstatus.c +++ b/pclsync/pstatus.c @@ -36,7 +36,7 @@ #include "plibs.h" #include "prunratelimit.h" #include "psettings.h" -#include "ptasks.h" +#include "ptask.h" #include "putil.h" #include #include diff --git a/pclsync/psyncer.c b/pclsync/psyncer.c index 9c628f9..6a6636b 100644 --- a/pclsync/psyncer.c +++ b/pclsync/psyncer.c @@ -48,7 +48,7 @@ #include "prun.h" #include "pstatus.h" #include "psyncer.h" -#include "ptasks.h" +#include "ptask.h" #include "ptree.h" #include "putil.h" #include @@ -275,7 +275,7 @@ void psync_add_folder_for_downloadsync(psync_syncid_t syncid, cfolderid = psync_get_number(row[0]); clfolderid = psync_create_local_folder_in_db(syncid, cfolderid, lfoiderid, name); - psync_task_create_local_folder(syncid, cfolderid, clfolderid); + ptask_ldir_mk(syncid, cfolderid, clfolderid); psync_add_folder_for_downloadsync(syncid, synctype, cfolderid, clfolderid /*, path*/); } @@ -287,7 +287,7 @@ void psync_add_folder_for_downloadsync(psync_syncid_t syncid, name = psync_get_string(row[1]); if (psync_is_name_to_ignore(name)) continue; - psync_task_download_file_silent(syncid, psync_get_number(row[0]), lfoiderid, + ptask_download_q(syncid, psync_get_number(row[0]), lfoiderid, name); } psync_sql_free_result(res); diff --git a/pclsync/psynclib.c b/pclsync/psynclib.c index f54ee77..73870b3 100644 --- a/pclsync/psynclib.c +++ b/pclsync/psynclib.c @@ -31,28 +31,27 @@ #include #include +#include +#include +#include + #include #include #include #include #include -#include -#include #include "papi.h" -#include "pasyncnet.h" #include "pbusinessaccount.h" #include "pcache.h" #include "pcallbacks.h" #include "pcloudcrypto.h" -#include "pfile.h" #include "pcontacts.h" #include "pdevice.h" #include "pdevice_monitor.h" #include "pdiff.h" -#include "ppath.h" #include "pdownload.h" -#include "psys.h" +#include "pfile.h" #include "pfileops.h" #include "pfolder.h" #include "pfsfolder.h" @@ -63,12 +62,11 @@ #include "pmemlock.h" #include "pnetlibs.h" #include "pnotifications.h" -#include "putil.h" - #include "poverlay.h" #include "pp2p.h" #include "ppagecache.h" #include "ppassword.h" +#include "ppath.h" #include "ppathstatus.h" #include "prun.h" #include "pscanner.h" @@ -78,13 +76,14 @@ #include "pstatus.h" #include "psyncer.h" #include "psynclib.h" +#include "psys.h" +#include "ptask.h" #include "ptimer.h" #include "ptools.h" #include "publiclinks.h" #include "pupload.h" -#include -#include -#include +#include "putil.h" + #define BUGHUNT 0 @@ -356,7 +355,7 @@ void psync_destroy() { psync_fs_stop(); psync_terminate_status_waiters(); psync_send_status_update(); - psync_async_stop(); + ptask_stop_async(); psync_timer_wake(); psync_timer_notify_exception(); psync_sql_sync(); @@ -447,7 +446,7 @@ void psync_logout2(uint32_t auth_status, int doinvauth) { psync_fs_pause_until_login(); psync_stop_all_download(); psync_stop_all_upload(); - psync_async_stop(); + ptask_stop_async(); psync_cache_clean_all(); psync_set_apiserver(PSYNC_API_HOST, PSYNC_LOCATIONID_DEFAULT); psync_restart_localscan(); diff --git a/pclsync/pasyncnet.c b/pclsync/ptask.c similarity index 53% rename from pclsync/pasyncnet.c rename to pclsync/ptask.c index 7202577..094cad7 100644 --- a/pclsync/pasyncnet.c +++ b/pclsync/ptask.c @@ -1,61 +1,63 @@ /* - Copyright (c) 2015 Anton Titov. + Copyright (c) 2013-2014 Anton Titov. - Copyright (c) 2015 pCloud Ltd. All rights reserved. + Copyright (c) 2013-2014 pCloud Ltd. All rights reserved. - Redistribution and use in source and binary forms, with or without - modification, are permitted provided that the following conditions - are met: Redistributions of source code must retain the above - copyright notice, this list of conditions and the following - disclaimer. Redistributions in binary form must reproduce the above - copyright notice, this list of conditions and the following - disclaimer in the documentation and/or other materials provided with - the distribution. Neither the name of pCloud Ltd nor the names of - its contributors may be used to endorse or promote products derived - from this software without specific prior written permission. + Redistribution and use in source and binary forms, with or without + modification, are permitted provided that the following conditions + are met: Redistributions of source code must retain the above + copyright notice, this list of conditions and the following + disclaimer. Redistributions in binary form must reproduce the + above copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. Neither the name of pCloud Ltd nor the + names of its contributors may be used to endorse or promote + products derived from this software without specific prior written + permission. - THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR - A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL pCloud Ltd BE - LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR - BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, - WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE - OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, - EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS + FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL pCloud + Ltd BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, + EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, + PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR + PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY + OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE + USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH + DAMAGE. */ #include #include +#include #include #include #include #include -#include #include +#include #include "papi.h" -#include "pasyncnet.h" -#include "pfile.h" +#include "pcallbacks.h" #include "pcompression.h" +#include "pdownload.h" +#include "pfile.h" #include "plibs.h" #include "pnetlibs.h" +#include "ppathstatus.h" #include "prun.h" #include "psettings.h" #include "pssl.h" +#include "pstatus.h" #include "psys.h" +#include "ptask.h" #include "ptree.h" -#include +#include "pupload.h" -#define TASK_WITH_HEADER(t1, t2) \ - typedef struct { \ - task_header_t head; \ - t2 task; \ - } t1 -#define get_len(t) (sizeof(t) - offsetof(t, task)) +#define get_len(t) (sizeof(t) - offsetof(t, request)) #define TASK_TYPE_EXIT 0 #define TASK_TYPE_FILE_DWL 1 @@ -65,17 +67,32 @@ #define STREAM_HEADER_LEN 6 // 4 bytes stream id, 2 bytes length +#define CHECK_LEN(l) \ + do { \ + if (unlikely(len != l)) { \ + debug(D_BUG, \ + "wrong size for packet of type %u, expected size %u but got %u", \ + (unsigned)type, (unsigned)l, (unsigned)len); \ + return 2; \ + } \ + } while (0) + typedef struct { uint32_t type; uint32_t len; -} task_header_t; +} hdr_t; typedef struct { psync_fileid_t fileid; const char *localpath; psync_async_callback_t cb; void *cbext; -} task_file_download_t; +} download_req_t; + +typedef struct { + hdr_t head; + download_req_t request; +} download_task_t; typedef struct { psync_fileid_t fileid; @@ -84,7 +101,12 @@ typedef struct { char sha1hex[PSYNC_SHA1_DIGEST_HEXLEN]; psync_async_callback_t cb; void *cbext; -} task_file_download_if_not_mod_t; +} download_needed_req_t; + +typedef struct { + hdr_t head; + download_needed_req_t request; +} download_needed_task_t; typedef struct { uint32_t error; @@ -95,15 +117,15 @@ typedef struct { uint64_t oldhash; uint64_t oldmtime; unsigned char sha1hex[PSYNC_SHA1_DIGEST_HEXLEN]; -} task_file_download_resp_t; +} download_rsp_t; -typedef struct _async_thread_params_t { +typedef struct _async_params_t { psync_deflate_t *enc; psync_deflate_t *dec; psock_t *api; psync_tree *streams; uint64_t datapendingsince; - int (*process_buf)(struct _async_thread_params_t *); + int (*process_buf)(struct _async_params_t *); char *curreadbuff; int privsock; uint32_t curreadbuffrem; @@ -112,7 +134,7 @@ typedef struct _async_thread_params_t { uint32_t laststreamid; uint32_t pendingrequests; char buffer[64 * 1024]; -} async_thread_params_t; +} async_params_t; typedef struct _stream_t { psync_tree tree; @@ -121,7 +143,7 @@ typedef struct _stream_t { psync_async_callback_t cb; void (*free)(struct _stream_t *, uint32_t); void *cbext; - int (*process_data)(struct _stream_t *, async_thread_params_t *, const char *, + int (*process_data)(struct _stream_t *, async_params_t *, const char *, uint32_t); } stream_t; @@ -136,17 +158,54 @@ typedef struct { psync_sha1_ctx sha1ctx; uint64_t remsize; int fd; -} file_download_add_t; +} download_context_t; -TASK_WITH_HEADER(task_hdr_file_download_t, task_file_download_t); -TASK_WITH_HEADER(task_hdr_file_download_if_not_mod_t, - task_file_download_if_not_mod_t); - -static pthread_mutex_t amutex = PTHREAD_MUTEX_INITIALIZER; +static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; static int running = 0; -static int at_sock = INVALID_SOCKET; +static int sockd = INVALID_SOCKET; -static int send_pending_data(async_thread_params_t *prms) { +static int data_pending_send(async_params_t *prms); +static int handle_command_data(async_params_t *prms, char *data, uint32_t type, uint32_t len); +static int sock_readall(int sock, void *buff, size_t len); +static int sock_writeall(int sock, const void *buff, size_t len); +static int stream_process_header(async_params_t *prms); +static stream_t *stream_create(async_params_t *prms, size_t addsize); +static void stream_destroy(stream_t *s); +static void stream_setup_header(async_params_t *prms); + +static int data_send(async_params_t *prms, const void *data, int len) { + int wr; + while (len) { + wr = psync_deflate_write(prms->enc, data, len, PSYNC_DEFLATE_NOFLUSH); + if (wr > 0) { + len -= wr; + data = (const char *)data + wr; + } else if (wr != PSYNC_DEFLATE_FULL) { + debug(D_ERROR, "write to deflate compressor of %d bytes returned %d", len, + wr); + return -1; + } + if (data_pending_send(prms)) + return -1; + } + if (!prms->pendingrequests) + prms->datapendingsince = psys_time_milliseconds(); + prms->pendingrequests++; + return 0; +} + +static int data_pending_flush(async_params_t *prms) { + int ret; + ret = psync_deflate_write(prms->enc, "", 0, PSYNC_DEFLATE_FLUSH); + if (ret != 0) { + debug(D_WARNING, "psync_deflate_write returned %d when flushing", ret); + return -1; + } + prms->pendingrequests = 0; + return data_pending_send(prms); +} + +static int data_pending_send(async_params_t *prms) { char buff[4096]; int ret; while (1) { @@ -166,111 +225,9 @@ static int send_pending_data(async_thread_params_t *prms) { } } -static int flush_pending_data(async_thread_params_t *prms) { - int ret; - ret = psync_deflate_write(prms->enc, "", 0, PSYNC_DEFLATE_FLUSH); - if (ret != 0) { - debug(D_WARNING, "psync_deflate_write returned %d when flushing", ret); - return -1; - } - prms->pendingrequests = 0; - return send_pending_data(prms); -} - -static int socket_t_readall(int sock, void *buff, size_t len) { - ssize_t rd; - while (len) { - if (psock_wait_read_timeout(sock)) - return -1; - rd = read(sock, buff, len); - if (rd > 0) { - len -= rd; - buff = (char *)buff + rd; - } else if (rd == PSYNC_SOCKET_ERROR && - (errno == EINTR || errno == EAGAIN || - errno == EWOULDBLOCK)) - continue; - else { - debug(D_WARNING, "read from socket of %lu bytes returned %ld, errno %d", - (unsigned long)len, (long)rd, (int)errno); - return -1; - } - } - return 0; -} - -static int socket_t_writeall(int sock, const void *buff, - size_t len) { - ssize_t wr; - while (len) { - if (psock_wait_write_timeout(sock)) - return -1; - wr = write(sock, buff, len); - if (wr > 0) { - len -= wr; - buff = (const char *)buff + wr; - } else if (wr == PSYNC_SOCKET_ERROR && - (errno == EINTR || errno == EAGAIN || - errno == EWOULDBLOCK)) - continue; - else { - debug(D_WARNING, "write to socket of %lu bytes returned %ld, errno %d", - (unsigned long)len, (long)wr, (int)errno); - return -1; - } - } - return 0; -} - -static stream_t *create_stream(async_thread_params_t *prms, size_t addsize) { - psync_tree *parent; - stream_t *ret; - ret = (stream_t *)psync_malloc(sizeof(stream_t) + addsize); - ret->streamid = ++prms->laststreamid; - ret->flags = 0; - ret->free = NULL; - parent = psync_tree_get_last(prms->streams); - if (parent) - parent->right = &ret->tree; - else - prms->streams = &ret->tree; - psync_tree_added_at(&prms->streams, parent, &ret->tree); - return ret; -} - -static int send_data(async_thread_params_t *prms, const void *data, int len) { - int wr; - while (len) { - wr = psync_deflate_write(prms->enc, data, len, PSYNC_DEFLATE_NOFLUSH); - if (wr > 0) { - len -= wr; - data = (const char *)data + wr; - } else if (wr != PSYNC_DEFLATE_FULL) { - debug(D_ERROR, "write to deflate compressor of %d bytes returned %d", len, - wr); - return -1; - } - if (send_pending_data(prms)) - return -1; - } - if (!prms->pendingrequests) - prms->datapendingsince = psys_time_milliseconds(); - prms->pendingrequests++; - return 0; -} - -static void close_stream(stream_t *s, async_thread_params_t *prms, - uint32_t error) { - debug(D_NOTICE, "closing stream %u", (unsigned)s->streamid); - psync_tree_del(&prms->streams, &s->tree); - if (s->free) - s->free(s, error); - psync_free(s); -} - -static void file_download_free(stream_t *s, uint32_t error) { - file_download_add_t *fda; - fda = (file_download_add_t *)(s + 1); +static void download_free(stream_t *s, uint32_t error) { + download_context_t *fda; + fda = (download_context_t *)(s + 1); if (fda->fd != INVALID_HANDLE_VALUE) { pfile_close(fda->fd); if (error) @@ -278,8 +235,8 @@ static void file_download_free(stream_t *s, uint32_t error) { } } -static int file_download_send_error(stream_t *s, async_thread_params_t *prms, - file_download_add_t *fda, uint32_t error, +static int download_send_error(stream_t *s, async_params_t *prms, + download_context_t *fda, uint32_t error, uint32_t errorflags) { psync_async_result_t r; if (error) @@ -293,11 +250,16 @@ static int file_download_send_error(stream_t *s, async_thread_params_t *prms, r.file.hash = fda->hash; memcpy(r.file.sha1hex, fda->sha1hex, PSYNC_SHA1_DIGEST_HEXLEN); s->cb(s->cbext, &r); - close_stream(s, prms, error); + + debug(D_NOTICE, "closing stream %u", (unsigned)s->streamid); + psync_tree_del(&prms->streams, &s->tree); + if (s->free) + s->free(s, error); + psync_free(s); return 0; } -static int file_download_checksum(file_download_add_t *fda) { +static int download_checksum(download_context_t *fda) { unsigned char sha1b[PSYNC_SHA1_DIGEST_LEN], sha1h[PSYNC_SHA1_DIGEST_HEXLEN]; psync_sha1_final(sha1b, &fda->sha1ctx); psync_binhex(sha1h, sha1b, PSYNC_SHA1_DIGEST_LEN); @@ -310,19 +272,19 @@ static int file_download_checksum(file_download_add_t *fda) { return 0; } -static int process_file_download_data(stream_t *s, async_thread_params_t *prms, +static int download_process_data(stream_t *s, async_params_t *prms, const char *buff, uint32_t datalen) { - file_download_add_t *fda; + download_context_t *fda; ssize_t wr; int err; - fda = (file_download_add_t *)(s + 1); + fda = (download_context_t *)(s + 1); if (datalen > fda->remsize) { debug(D_ERROR, "got packed of size %u for stream %u file %s when the remaining data " "is %lu", (unsigned)datalen, (unsigned)s->streamid, fda->localpath, (unsigned long)fda->remsize); - file_download_send_error(s, prms, fda, PSYNC_ASYNC_ERROR_NET, + download_send_error(s, prms, fda, PSYNC_ASYNC_ERROR_NET, PSYNC_ASYNC_ERR_FLAG_RETRY_AS_IS); return -1; } @@ -335,7 +297,7 @@ static int process_file_download_data(stream_t *s, async_thread_params_t *prms, err = (int)errno; debug(D_WARNING, "writing to file %s failed, errno %d", fda->localpath, err); - return file_download_send_error( + return download_send_error( s, prms, fda, err == ENOSPC ? PSYNC_ASYNC_ERROR_DISK_FULL : PSYNC_ASYNC_ERROR_IO, 0); @@ -344,34 +306,34 @@ static int process_file_download_data(stream_t *s, async_thread_params_t *prms, buff += wr; } if (fda->remsize == 0) { - if (file_download_checksum(fda)) - return file_download_send_error(s, prms, fda, PSYNC_ASYNC_ERROR_CHECKSUM, + if (download_checksum(fda)) + return download_send_error(s, prms, fda, PSYNC_ASYNC_ERROR_CHECKSUM, 0); else - return file_download_send_error(s, prms, fda, 0, 0); + return download_send_error(s, prms, fda, 0, 0); } else return 0; } -static int process_file_download_headers(stream_t *s, - async_thread_params_t *prms, +static int download_process_headers(stream_t *s, + async_params_t *prms, const char *buff, uint32_t datalen) { - task_file_download_resp_t r; - file_download_add_t *fda; + download_rsp_t r; + download_context_t *fda; psync_sql_res *res; - if (unlikely(datalen < sizeof(task_file_download_resp_t))) { + if (unlikely(datalen < sizeof(download_rsp_t))) { debug(D_ERROR, "got packet of size %u while expecting at least %u, disconnecting", - (unsigned)datalen, (unsigned)sizeof(task_file_download_resp_t)); + (unsigned)datalen, (unsigned)sizeof(download_rsp_t)); return -1; } - memcpy(&r, buff, sizeof(task_file_download_resp_t)); - fda = (file_download_add_t *)(s + 1); + memcpy(&r, buff, sizeof(download_rsp_t)); + fda = (download_context_t *)(s + 1); fda->size = r.size; fda->hash = r.hash; memcpy(fda->sha1hex, r.sha1hex, PSYNC_SHA1_DIGEST_HEXLEN); if (r.error) - return file_download_send_error(s, prms, fda, r.error + 100, r.errorflags); + return download_send_error(s, prms, fda, r.error + 100, r.errorflags); debug(D_NOTICE, "got headers for file %s size %" PRIu64 " hash %" PRIu64 " sha1 %.40s", @@ -412,118 +374,27 @@ static int process_file_download_headers(stream_t *s, if (fda->fd == INVALID_HANDLE_VALUE) { debug(D_WARNING, "could not open file %s, errno %d", fda->localpath, (int)errno); - return file_download_send_error(s, prms, fda, PSYNC_ASYNC_ERROR_FILE, 0); + return download_send_error(s, prms, fda, PSYNC_ASYNC_ERROR_FILE, 0); } fda->remsize = fda->size; if (!fda->remsize) - return file_download_send_error(s, prms, fda, 0, 0); - s->process_data = process_file_download_data; + return download_send_error(s, prms, fda, 0, 0); + s->process_data = download_process_data; psync_sha1_init(&fda->sha1ctx); - if (datalen > sizeof(task_file_download_resp_t)) - return process_file_download_data( - s, prms, buff + sizeof(task_file_download_resp_t), - datalen - sizeof(task_file_download_resp_t)); + if (datalen > sizeof(download_rsp_t)) + return download_process_data( + s, prms, buff + sizeof(download_rsp_t), + datalen - sizeof(download_rsp_t)); else return 0; } -static int handle_file_download(async_thread_params_t *prms, - task_file_download_t *dwl) { - char buff[256]; - stream_t *s; - file_download_add_t *fda; - int len; - s = create_stream(prms, sizeof(file_download_add_t)); - fda = (file_download_add_t *)(s + 1); - fda->fileid = dwl->fileid; - s->free = file_download_free; - s->cb = dwl->cb; - s->cbext = dwl->cbext; - s->process_data = process_file_download_headers; - fda->localpath = dwl->localpath; - fda->fd = INVALID_HANDLE_VALUE; - len = psync_slprintf(buff, sizeof(buff), - "act=dwl,strm=%" PRIu64 ",fileid=%" PRIu64 "\n", - (uint64_t)s->streamid, (uint64_t)dwl->fileid); - if (send_data(prms, buff, len)) { - debug(D_WARNING, "failed to send request for fileid %lu", - (unsigned long)dwl->fileid); - return -1; - } - s->flags |= STREAM_FLAG_ACTIVE; - debug(D_NOTICE, "requested data of fileid %lu to be saved in %s", - (unsigned long)dwl->fileid, dwl->localpath); - return 0; -} - -static int handle_file_download_nm(async_thread_params_t *prms, - task_file_download_if_not_mod_t *dwl) { - char buff[256]; - stream_t *s; - file_download_add_t *fda; - int len; - s = create_stream(prms, sizeof(file_download_add_t)); - fda = (file_download_add_t *)(s + 1); - fda->fileid = dwl->fileid; - s->free = file_download_free; - s->cb = dwl->cb; - s->cbext = dwl->cbext; - s->process_data = process_file_download_headers; - fda->osize = dwl->size; - memcpy(fda->osha1hex, dwl->sha1hex, PSYNC_SHA1_DIGEST_HEXLEN); - fda->localpath = dwl->localpath; - fda->fd = INVALID_HANDLE_VALUE; - len = psync_slprintf( - buff, sizeof(buff), - "act=dwlnm,strm=%" PRIu64 ",fileid=%" PRIu64 ",sha1=%.40s\n", - (uint64_t)s->streamid, (uint64_t)dwl->fileid, dwl->sha1hex); - if (send_data(prms, buff, len)) { - debug(D_WARNING, "failed to send request for fileid %lu", - (unsigned long)dwl->fileid); - return -1; - } - s->flags |= STREAM_FLAG_ACTIVE; - debug(D_NOTICE, "requested data of fileid %lu to be saved in %s", - (unsigned long)dwl->fileid, dwl->localpath); - return 0; -} - -#define CHECK_LEN(l) \ - do { \ - if (unlikely(len != l)) { \ - debug(D_BUG, \ - "wrong size for packet of type %u, expected size %u but got %u", \ - (unsigned)type, (unsigned)l, (unsigned)len); \ - return 2; \ - } \ - } while (0) - -static int handle_command_data(async_thread_params_t *prms, char *data, - uint32_t type, uint32_t len) { - switch (type) { - case TASK_TYPE_EXIT: - CHECK_LEN(0); - debug(D_NOTICE, "exiting"); - return -1; - case TASK_TYPE_FILE_DWL: - CHECK_LEN(sizeof(task_file_download_t)); - return handle_file_download(prms, (task_file_download_t *)data); - case TASK_TYPE_FILE_DWL_NM: - CHECK_LEN(sizeof(task_file_download_if_not_mod_t)); - return handle_file_download_nm(prms, - (task_file_download_if_not_mod_t *)data); - default: - debug(D_BUG, "got packet of unknown type %u", (unsigned)type); - return 1; - } -} - -static int handle_command(async_thread_params_t *prms) { +static int handle_command(async_params_t *prms) { char buff[4096]; - task_header_t hdr; + hdr_t hdr; int ret; unsigned char r; - if (socket_t_readall(prms->privsock, &hdr, sizeof(hdr))) { + if (sock_readall(prms->privsock, &hdr, sizeof(hdr))) { debug(D_WARNING, "could not read header from socket pair"); return -1; } @@ -533,7 +404,7 @@ static int handle_command(async_thread_params_t *prms) { (unsigned)hdr.len, (unsigned)sizeof(buff)); return -1; } - if (socket_t_readall(prms->privsock, buff, hdr.len)) { + if (sock_readall(prms->privsock, buff, hdr.len)) { debug(D_WARNING, "could not read %u bytes of data from socket pair", (unsigned)hdr.len); return -1; @@ -546,14 +417,94 @@ static int handle_command(async_thread_params_t *prms) { r = (unsigned char)ret; ret = 0; } - if (socket_t_writeall(prms->privsock, &r, sizeof(r))) { + if (sock_writeall(prms->privsock, &r, sizeof(r))) { debug(D_WARNING, "failed to write response to socket pair"); return -1; } return ret; } -static int handle_decompressed_data(async_thread_params_t *prms) { +static int handle_download(async_params_t *prms, + download_req_t *dwl) { + char buff[256]; + stream_t *s; + download_context_t *fda; + int len; + s = stream_create(prms, sizeof(download_context_t)); + fda = (download_context_t *)(s + 1); + fda->fileid = dwl->fileid; + s->free = download_free; + s->cb = dwl->cb; + s->cbext = dwl->cbext; + s->process_data = download_process_headers; + fda->localpath = dwl->localpath; + fda->fd = INVALID_HANDLE_VALUE; + len = psync_slprintf(buff, sizeof(buff), + "act=dwl,strm=%" PRIu64 ",fileid=%" PRIu64 "\n", + (uint64_t)s->streamid, (uint64_t)dwl->fileid); + if (data_send(prms, buff, len)) { + debug(D_WARNING, "failed to send request for fileid %lu", + (unsigned long)dwl->fileid); + return -1; + } + s->flags |= STREAM_FLAG_ACTIVE; + debug(D_NOTICE, "requested data of fileid %lu to be saved in %s", + (unsigned long)dwl->fileid, dwl->localpath); + return 0; +} + +static int handle_download_nm(async_params_t *prms, + download_needed_req_t *dwl) { + char buff[256]; + stream_t *s; + download_context_t *fda; + int len; + s = stream_create(prms, sizeof(download_context_t)); + fda = (download_context_t *)(s + 1); + fda->fileid = dwl->fileid; + s->free = download_free; + s->cb = dwl->cb; + s->cbext = dwl->cbext; + s->process_data = download_process_headers; + fda->osize = dwl->size; + memcpy(fda->osha1hex, dwl->sha1hex, PSYNC_SHA1_DIGEST_HEXLEN); + fda->localpath = dwl->localpath; + fda->fd = INVALID_HANDLE_VALUE; + len = psync_slprintf( + buff, sizeof(buff), + "act=dwlnm,strm=%" PRIu64 ",fileid=%" PRIu64 ",sha1=%.40s\n", + (uint64_t)s->streamid, (uint64_t)dwl->fileid, dwl->sha1hex); + if (data_send(prms, buff, len)) { + debug(D_WARNING, "failed to send request for fileid %lu", + (unsigned long)dwl->fileid); + return -1; + } + s->flags |= STREAM_FLAG_ACTIVE; + debug(D_NOTICE, "requested data of fileid %lu to be saved in %s", + (unsigned long)dwl->fileid, dwl->localpath); + return 0; +} + +static int handle_command_data(async_params_t *prms, char *data, uint32_t type, uint32_t len) { + switch (type) { + case TASK_TYPE_EXIT: + CHECK_LEN(0); + debug(D_NOTICE, "exiting"); + return -1; + case TASK_TYPE_FILE_DWL: + CHECK_LEN(sizeof(download_req_t)); + return handle_download(prms, (download_req_t *)data); + case TASK_TYPE_FILE_DWL_NM: + CHECK_LEN(sizeof(download_needed_req_t)); + return handle_download_nm(prms, + (download_needed_req_t *)data); + default: + debug(D_BUG, "got packet of unknown type %u", (unsigned)type); + return 1; + } +} + +static int handle_decompressed_data(async_params_t *prms) { int rd; while (1) { rd = psync_deflate_read(prms->dec, prms->curreadbuff, prms->curreadbuffrem); @@ -571,7 +522,7 @@ static int handle_decompressed_data(async_thread_params_t *prms) { } } -static int handle_incoming_data(async_thread_params_t *prms) { +static int handle_incoming_data(async_params_t *prms) { char buff[4096]; char *ptr; int rdsock, wrdecomp; @@ -601,77 +552,19 @@ static int handle_incoming_data(async_thread_params_t *prms) { } } -static void free_stream(stream_t *s) { - debug(D_NOTICE, "freeing unfinished stream %u", (unsigned)s->streamid); - if (s->flags & STREAM_FLAG_ACTIVE) { - psync_async_result_t ar; - memset(&ar, 0, sizeof(ar)); - ar.error = PSYNC_ASYNC_ERROR_NET; - ar.errorflags = PSYNC_ASYNC_ERR_FLAG_RETRY_AS_IS; - s->cb(s->cbext, &ar); - } - if (s->free) - s->free(s, PSYNC_ASYNC_ERROR_NET); - psync_free(s); -} - -static int process_stream_header(async_thread_params_t *prms); - -static void read_stream_header_setup(async_thread_params_t *prms) { - prms->curreadbuff = prms->buffer; - prms->curreadbuffrem = STREAM_HEADER_LEN; - prms->process_buf = process_stream_header; -} - -static int process_stream_data(async_thread_params_t *prms) { - psync_tree *tr; - stream_t *s; - int ret; - tr = prms->streams; - while (tr) { - s = psync_tree_element(tr, stream_t, tree); - if (prms->currentstreamid < s->streamid) - tr = tr->left; - else if (prms->currentstreamid > s->streamid) - tr = tr->right; - else { - ret = s->process_data(s, prms, prms->buffer, prms->curreadbufflen); - break; - } - } - if (!tr) { - debug(D_NOTICE, "throwing out %u bytes of data for unknown streamid %u", - (unsigned)prms->curreadbufflen, (unsigned)prms->currentstreamid); - ret = 0; - } - read_stream_header_setup(prms); - return ret; -} - -static int process_stream_header(async_thread_params_t *prms) { - uint32_t len; - memcpy(&prms->currentstreamid, prms->buffer, 4); - len = 0; - memcpy(&len, prms->buffer + 4, 2); - prms->curreadbuff = prms->buffer; - prms->curreadbufflen = prms->curreadbuffrem = len + 1; - prms->process_buf = process_stream_data; - return 0; -} - -static void psync_async_thread(void *ptr) { - async_thread_params_t *prms = (async_thread_params_t *)ptr; +static void proc_async_transfer(void *ptr) { + async_params_t *prms = (async_params_t *)ptr; int sel[2]; int ret; sel[0] = prms->api->sock; sel[1] = prms->privsock; - read_stream_header_setup(prms); + stream_setup_header(prms); while (1) { if (prms->pendingrequests) { if ((prms->pendingrequests >= PSYNC_ASYNC_MAX_GROUPED_REQUESTS || prms->datapendingsince + PSYNC_ASYNC_GROUP_REQUESTS_FOR < psys_time_milliseconds()) && - flush_pending_data(prms)) + data_pending_flush(prms)) break; ret = psock_select_in(sel, 2, PSYNC_ASYNC_GROUP_REQUESTS_FOR / 4); if (ret == -1) @@ -697,26 +590,26 @@ static void psync_async_thread(void *ptr) { // close prms->privsock before locking as there might be somebody who keeps // the mutex locked while waiting for us to reply close(prms->privsock); - pthread_mutex_lock(&amutex); - close(at_sock); - at_sock = INVALID_SOCKET; + pthread_mutex_lock(&mutex); + close(sockd); + sockd = INVALID_SOCKET; running--; - pthread_mutex_unlock(&amutex); + pthread_mutex_unlock(&mutex); psync_apipool_release_bad(prms->api); psync_deflate_destroy(prms->enc); psync_deflate_destroy(prms->dec); psync_tree_for_each_element_call_safe(prms->streams, stream_t, tree, - free_stream); + stream_destroy); psync_free(prms); } -static int psync_async_start_thread_locked() { +static int proc_start_async_transfer() { /* If some form of protocol version negotiation is to be performed, here is * the place to pass any needed parameters. The assumption will be that server * supports everything and clients inform the server what they support. */ binparam params[] = {PAPI_STR("auth", psync_my_auth), PAPI_STR("checksum", "sha1")}; - async_thread_params_t *tparams; + async_params_t *tparams; psync_deflate_t *enc, *dec; binresult *res; psock_t *api; @@ -763,14 +656,14 @@ static int psync_async_start_thread_locked() { debug(D_NOTICE, "psync_deflate_init() failed"); goto err3; } - tparams = psync_new(async_thread_params_t); - memset(tparams, 0, sizeof(async_thread_params_t)); + tparams = psync_new(async_params_t); + memset(tparams, 0, sizeof(async_params_t)); tparams->enc = enc; tparams->dec = dec; tparams->api = api; tparams->privsock = pair[1]; - at_sock = pair[0]; - prun_thread1("async transfer", psync_async_thread, tparams); + sockd = pair[0]; + prun_thread1("async transfer", proc_async_transfer, tparams); running++; return 0; err3: @@ -784,14 +677,130 @@ err0: return -1; } -static int psync_async_send_task_locked(const void *task, size_t len) { +static int sock_readall(int sock, void *buff, size_t len) { + ssize_t rd; + while (len) { + if (psock_wait_read_timeout(sock)) + return -1; + rd = read(sock, buff, len); + if (rd > 0) { + len -= rd; + buff = (char *)buff + rd; + } else if (rd == PSYNC_SOCKET_ERROR && + (errno == EINTR || errno == EAGAIN || + errno == EWOULDBLOCK)) + continue; + else { + debug(D_WARNING, "read from socket of %lu bytes returned %ld, errno %d", + (unsigned long)len, (long)rd, (int)errno); + return -1; + } + } + return 0; +} + +static int sock_writeall(int sock, const void *buff, size_t len) { + ssize_t wr; + while (len) { + if (psock_wait_write_timeout(sock)) + return -1; + wr = write(sock, buff, len); + if (wr > 0) { + len -= wr; + buff = (const char *)buff + wr; + } else if (wr == PSYNC_SOCKET_ERROR && + (errno == EINTR || errno == EAGAIN || + errno == EWOULDBLOCK)) + continue; + else { + debug(D_WARNING, "write to socket of %lu bytes returned %ld, errno %d", + (unsigned long)len, (long)wr, (int)errno); + return -1; + } + } + return 0; +} + +static stream_t *stream_create(async_params_t *prms, size_t addsize) { + psync_tree *parent; + stream_t *ret; + ret = (stream_t *)psync_malloc(sizeof(stream_t) + addsize); + ret->streamid = ++prms->laststreamid; + ret->flags = 0; + ret->free = NULL; + parent = psync_tree_get_last(prms->streams); + if (parent) + parent->right = &ret->tree; + else + prms->streams = &ret->tree; + psync_tree_added_at(&prms->streams, parent, &ret->tree); + return ret; +} + +static void stream_destroy(stream_t *s) { + debug(D_NOTICE, "freeing unfinished stream %u", (unsigned)s->streamid); + if (s->flags & STREAM_FLAG_ACTIVE) { + psync_async_result_t ar; + memset(&ar, 0, sizeof(ar)); + ar.error = PSYNC_ASYNC_ERROR_NET; + ar.errorflags = PSYNC_ASYNC_ERR_FLAG_RETRY_AS_IS; + s->cb(s->cbext, &ar); + } + if (s->free) + s->free(s, PSYNC_ASYNC_ERROR_NET); + psync_free(s); +} + +static int stream_process_data(async_params_t *prms) { + psync_tree *tr; + stream_t *s; + int ret; + tr = prms->streams; + while (tr) { + s = psync_tree_element(tr, stream_t, tree); + if (prms->currentstreamid < s->streamid) + tr = tr->left; + else if (prms->currentstreamid > s->streamid) + tr = tr->right; + else { + ret = s->process_data(s, prms, prms->buffer, prms->curreadbufflen); + break; + } + } + if (!tr) { + debug(D_NOTICE, "throwing out %u bytes of data for unknown streamid %u", + (unsigned)prms->curreadbufflen, (unsigned)prms->currentstreamid); + ret = 0; + } + stream_setup_header(prms); + return ret; +} + +static int stream_process_header(async_params_t *prms) { + uint32_t len; + memcpy(&prms->currentstreamid, prms->buffer, 4); + len = 0; + memcpy(&len, prms->buffer + 4, 2); + prms->curreadbuff = prms->buffer; + prms->curreadbufflen = prms->curreadbuffrem = len + 1; + prms->process_buf = stream_process_data; + return 0; +} + +static void stream_setup_header(async_params_t *prms) { + prms->curreadbuff = prms->buffer; + prms->curreadbuffrem = STREAM_HEADER_LEN; + prms->process_buf = stream_process_header; +} + +static int task_send_async_locked(const void *task, size_t len) { unsigned char ch; - if (socket_t_writeall(at_sock, task, len)) { + if (sock_writeall(sockd, task, len)) { debug(D_WARNING, "failed to write %lu bytes of task to socket", (unsigned long)len); return -1; } - if (socket_t_readall(at_sock, &ch, 1)) { + if (sock_readall(sockd, &ch, 1)) { debug(D_WARNING, "failed to read response from socket"); return -1; } @@ -803,55 +812,247 @@ static int psync_async_send_task_locked(const void *task, size_t len) { } } -static int psync_async_send_task(const void *task, size_t len) { +static int task_send_async(const void *task, size_t len) { int ret; - pthread_mutex_lock(&amutex); + pthread_mutex_lock(&mutex); if (running) - ret = psync_async_send_task_locked(task, len); + ret = task_send_async_locked(task, len); else { - ret = psync_async_start_thread_locked(); + ret = proc_start_async_transfer(); if (!ret) - ret = psync_async_send_task_locked(task, len); + ret = task_send_async_locked(task, len); } - pthread_mutex_unlock(&amutex); + pthread_mutex_unlock(&mutex); return ret; } -void psync_async_stop() { - task_header_t task; +void ptask_ldir_mk(psync_syncid_t syncid, + psync_folderid_t folderid, + psync_folderid_t localfolderid) { + psync_sql_res *res; + res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " + "localitemid) VALUES (?, ?, ?, ?)"); + psync_sql_bind_uint(res, 1, PSYNC_CREATE_LOCAL_FOLDER); + psync_sql_bind_uint(res, 2, syncid); + psync_sql_bind_uint(res, 3, folderid); + psync_sql_bind_uint(res, 4, localfolderid); + psync_sql_run_free(res); +} + +void ptask_ldir_rm(psync_syncid_t syncid, psync_folderid_t folderid, psync_folderid_t localfolderid, const char *remotepath) { + psync_sql_res *res; + res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " + "localitemid, name) VALUES (?, ?, ?, ?, ?)"); + psync_sql_bind_uint(res, 1, PSYNC_DELETE_LOCAL_FOLDER); + psync_sql_bind_uint(res, 2, syncid); + psync_sql_bind_uint(res, 3, folderid); + psync_sql_bind_uint(res, 4, localfolderid); + psync_sql_bind_string(res, 5, remotepath); + psync_sql_run_free(res); +} + +void ptask_ldir_rm_r(psync_syncid_t syncid, psync_folderid_t folderid, psync_folderid_t localfolderid) { + psync_sql_res *res; + res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " + "localitemid) VALUES (?, ?, ?, ?)"); + psync_sql_bind_uint(res, 1, PSYNC_DELREC_LOCAL_FOLDER); + psync_sql_bind_uint(res, 2, syncid); + psync_sql_bind_uint(res, 3, folderid); + psync_sql_bind_uint(res, 4, localfolderid); + psync_sql_run_free(res); +} + +void ptask_ldir_rename(psync_syncid_t syncid, psync_folderid_t folderid, psync_folderid_t localfolderid, psync_folderid_t newlocalparentfolderid, const char *newname) { + psync_sql_res *res; + res = psync_sql_prep_statement( + "INSERT INTO task (type, syncid, itemid, localitemid, newitemid, name) " + "VALUES (?, ?, ?, ?, ?, ?)"); + psync_sql_bind_uint(res, 1, PSYNC_RENAME_LOCAL_FOLDER); + psync_sql_bind_uint(res, 2, syncid); + psync_sql_bind_uint(res, 3, folderid); + psync_sql_bind_uint(res, 4, localfolderid); + psync_sql_bind_uint(res, 5, newlocalparentfolderid); + psync_sql_bind_string(res, 6, newname); + psync_sql_run_free(res); +} + +void ptask_download_q(psync_syncid_t syncid, psync_fileid_t fileid, psync_folderid_t localfolderid, const char *name) { + psync_sql_res *res; + res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " + "localitemid, name) VALUES (?, ?, ?, ?, ?)"); + psync_sql_bind_uint(res, 1, PSYNC_DOWNLOAD_FILE); + psync_sql_bind_uint(res, 2, syncid); + psync_sql_bind_uint(res, 3, fileid); + psync_sql_bind_uint(res, 4, localfolderid); + psync_sql_bind_string(res, 5, name); + psync_path_status_sync_folder_task_added_locked(syncid, localfolderid); + psync_sql_run_free(res); +} + +void ptask_download(psync_syncid_t syncid, psync_fileid_t fileid, psync_folderid_t localfolderid, const char *name) { + ptask_download_q(syncid, fileid, localfolderid, name); + psync_wake_download(); + psync_status_recalc_to_download(); + psync_send_status_update(); +} + +void ptask_lfile_rename(psync_syncid_t oldsyncid, psync_syncid_t newsyncid, psync_fileid_t fileid, psync_folderid_t oldlocalfolderid, psync_folderid_t newlocalfolderid, const char *newname) { + psync_sql_res *res; + res = psync_sql_prep_statement( + "INSERT INTO task (type, syncid, newsyncid, itemid, localitemid, " + "newitemid, name) VALUES (?, ?, ?, ?, ?, ?, ?)"); + psync_sql_bind_uint(res, 1, PSYNC_RENAME_LOCAL_FILE); + psync_sql_bind_uint(res, 2, oldsyncid); + psync_sql_bind_uint(res, 3, newsyncid); + psync_sql_bind_uint(res, 4, fileid); + psync_sql_bind_uint(res, 5, oldlocalfolderid); + psync_sql_bind_uint(res, 6, newlocalfolderid); + psync_sql_bind_string(res, 7, newname); + psync_sql_run_free(res); +} + +void ptask_lfile_rm(psync_fileid_t fileid, const char *remotepath) { + psync_sql_res *res; + res = psync_sql_prep_statement( + "INSERT INTO task (type, itemid, localitemid, name) VALUES (?, ?, 0, ?)"); + psync_sql_bind_uint(res, 1, PSYNC_DELETE_LOCAL_FILE); + psync_sql_bind_uint(res, 2, fileid); + psync_sql_bind_string(res, 3, remotepath); + psync_sql_run_free(res); +} + +void ptask_lfile_rm_id(psync_syncid_t syncid, psync_fileid_t fileid, const char *remotepath) { + psync_sql_res *res; + res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " + "localitemid, name) VALUES (?, ?, ?, 0, ?)"); + psync_sql_bind_uint(res, 1, PSYNC_DELETE_LOCAL_FILE); + psync_sql_bind_uint(res, 2, syncid); + psync_sql_bind_uint(res, 3, fileid); + psync_sql_bind_string(res, 4, remotepath); + psync_sql_run_free(res); +} + +void ptask_rdir_mk(psync_syncid_t syncid, psync_folderid_t localfolderid, const char *name) { + psync_sql_res *res; + res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " + "localitemid, name) VALUES (?, ?, ?, ?, ?)"); + psync_sql_bind_uint(res, 1, PSYNC_CREATE_REMOTE_FOLDER); + psync_sql_bind_uint(res, 2, syncid); + psync_sql_bind_uint(res, 3, 0); + psync_sql_bind_uint(res, 4, localfolderid); + psync_sql_bind_string(res, 5, name); + psync_sql_run_free(res); +} + +void ptask_upload_q(psync_syncid_t syncid, psync_fileid_t localfileid, const char *name) { + psync_sql_res *res; + res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " + "localitemid, name) VALUES (?, ?, ?, ?, ?)"); + psync_sql_bind_uint(res, 1, PSYNC_UPLOAD_FILE); + psync_sql_bind_uint(res, 2, syncid); + psync_sql_bind_uint(res, 3, 0); + psync_sql_bind_uint(res, 4, localfileid); + psync_sql_bind_string(res, 5, name); + psync_sql_run_free(res); +} + +void ptask_upload(psync_syncid_t syncid, psync_fileid_t localfileid, const char *name) { + psync_sql_res *res; + res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " + "localitemid, name) VALUES (?, ?, ?, ?, ?)"); + psync_sql_bind_uint(res, 1, PSYNC_UPLOAD_FILE); + psync_sql_bind_uint(res, 2, syncid); + psync_sql_bind_uint(res, 3, 0); + psync_sql_bind_uint(res, 4, localfileid); + psync_sql_bind_string(res, 5, name); + psync_sql_run_free(res); + psync_wake_upload(); + psync_status_recalc_to_upload_async(); +} + +void ptask_rfile_rename(psync_syncid_t oldsyncid, psync_syncid_t newsyncid, psync_fileid_t localfileid, psync_folderid_t newlocalparentfolderid, const char *newname) { + psync_sql_res *res; + res = psync_sql_prep_statement( + "INSERT INTO task (type, syncid, newsyncid, localitemid, newitemid, " + "name, itemid) VALUES (?, ?, ?, ?, ?, ?, 0)"); + psync_sql_bind_uint(res, 1, PSYNC_RENAME_REMOTE_FILE); + psync_sql_bind_uint(res, 2, oldsyncid); + psync_sql_bind_uint(res, 3, newsyncid); + psync_sql_bind_uint(res, 4, localfileid); + psync_sql_bind_uint(res, 5, newlocalparentfolderid); + psync_sql_bind_string(res, 6, newname); + psync_sql_run_free(res); +} + +void ptask_rdir_rename(psync_syncid_t oldsyncid, psync_syncid_t newsyncid, psync_fileid_t localfileid, psync_folderid_t newlocalparentfolderid, const char *newname) { + psync_sql_res *res; + res = psync_sql_prep_statement( + "INSERT INTO task (type, syncid, newsyncid, localitemid, newitemid, " + "name, itemid) VALUES (?, ?, ?, ?, ?, ?, 0)"); + psync_sql_bind_uint(res, 1, PSYNC_RENAME_REMOTE_FOLDER); + psync_sql_bind_uint(res, 2, oldsyncid); + psync_sql_bind_uint(res, 3, newsyncid); + psync_sql_bind_uint(res, 4, localfileid); + psync_sql_bind_uint(res, 5, newlocalparentfolderid); + psync_sql_bind_string(res, 6, newname); + psync_sql_run_free(res); +} + +void ptask_rfile_rm(psync_syncid_t syncid, psync_fileid_t fileid) { + psync_sql_res *res; + res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " + "localitemid) VALUES (?, ?, ?, 0)"); + psync_sql_bind_uint(res, 1, PSYNC_DELETE_REMOTE_FILE); + psync_sql_bind_uint(res, 2, syncid); + psync_sql_bind_uint(res, 3, fileid); + psync_sql_run_free(res); +} + +void ptask_rdir_rm(psync_syncid_t syncid, psync_folderid_t folderid) { + psync_sql_res *res; + res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " + "localitemid) VALUES (?, ?, ?, 0)"); + psync_sql_bind_uint(res, 1, PSYNC_DELETE_REMOTE_FILE); + psync_sql_bind_uint(res, 2, syncid); + psync_sql_bind_uint(res, 3, folderid); + psync_sql_run_free(res); +} + +void ptask_stop_async() { + hdr_t task; task.type = TASK_TYPE_EXIT; task.len = 0; - pthread_mutex_lock(&amutex); + pthread_mutex_lock(&mutex); if (running) - psync_async_send_task_locked(&task, sizeof(task)); - pthread_mutex_unlock(&amutex); + task_send_async_locked(&task, sizeof(task)); + pthread_mutex_unlock(&mutex); } -int psync_async_download_file(psync_fileid_t fileid, const char *localpath, +int ptask_download_async(psync_fileid_t fileid, const char *localpath, psync_async_callback_t cb, void *cbext) { - task_hdr_file_download_t task; + download_task_t task; task.head.type = TASK_TYPE_FILE_DWL; - task.head.len = get_len(task_hdr_file_download_t); - task.task.fileid = fileid; - task.task.localpath = localpath; - task.task.cb = cb; - task.task.cbext = cbext; - return psync_async_send_task(&task, sizeof(task)); + task.head.len = get_len(download_task_t); + task.request.fileid = fileid; + task.request.localpath = localpath; + task.request.cb = cb; + task.request.cbext = cbext; + return task_send_async(&task, sizeof(task)); } -int psync_async_download_file_if_changed(psync_fileid_t fileid, +int ptask_download_needed_async(psync_fileid_t fileid, const char *localpath, uint64_t size, const void *sha1hex, psync_async_callback_t cb, void *cbext) { - task_hdr_file_download_if_not_mod_t task; + download_needed_task_t task; task.head.type = TASK_TYPE_FILE_DWL_NM; - task.head.len = get_len(task_hdr_file_download_if_not_mod_t); - task.task.fileid = fileid; - task.task.localpath = localpath; - task.task.size = size; - memcpy(task.task.sha1hex, sha1hex, PSYNC_SHA1_DIGEST_HEXLEN); - task.task.cb = cb; - task.task.cbext = cbext; - return psync_async_send_task(&task, sizeof(task)); + task.head.len = get_len(download_needed_task_t); + task.request.fileid = fileid; + task.request.localpath = localpath; + task.request.size = size; + memcpy(task.request.sha1hex, sha1hex, PSYNC_SHA1_DIGEST_HEXLEN); + task.request.cb = cb; + task.request.cbext = cbext; + return task_send_async(&task, sizeof(task)); } diff --git a/pclsync/ptasks.h b/pclsync/ptask.h similarity index 55% rename from pclsync/ptasks.h rename to pclsync/ptask.h index 2196e84..838610f 100644 --- a/pclsync/ptasks.h +++ b/pclsync/ptask.h @@ -32,18 +32,29 @@ #ifndef _PSYNC_TASKS_H #define _PSYNC_TASKS_H -#include "pcompiler.h" #include "psynclib.h" +#define PSYNC_ASYNC_ERR_FLAG_PERM \ + 0x01 // the error is permanent(ish) and there is no reason to retry +#define PSYNC_ASYNC_ERR_FLAG_RETRY_AS_IS \ + 0x02 // same request may succeed in the future if retried as is +#define PSYNC_ASYNC_ERR_FLAG_SUCCESS \ + 0x04 // like no action performed because of no need - file already exists and + // so on + +#define PSYNC_ASYNC_ERROR_NET 1 +#define PSYNC_ASYNC_ERROR_FILE 2 +#define PSYNC_ASYNC_ERROR_DISK_FULL 3 +#define PSYNC_ASYNC_ERROR_IO 4 +#define PSYNC_ASYNC_ERROR_CHECKSUM 5 +#define PSYNC_SERVER_ERROR_TOO_BIG 102 +#define PSYNC_SERVER_ERROR_NOT_MOD 104 #define PSYNC_TASK_DOWNLOAD 0 #define PSYNC_TASK_UPLOAD 1 #define PSYNC_TASK_DWLUPL_MASK 1 - #define PSYNC_TASK_FOLDER 0 #define PSYNC_TASK_FILE 2 - #define PSYNC_TASK_TYPE_OFF 2 - #define PSYNC_TASK_TYPE_CREATE 0 #define PSYNC_TASK_TYPE_DELETE 1 #define PSYNC_TASK_TYPE_DELREC 2 @@ -94,65 +105,53 @@ ((PSYNC_TASK_TYPE_DELREC << PSYNC_TASK_TYPE_OFF) + PSYNC_TASK_FOLDER + \ PSYNC_TASK_UPLOAD) -void psync_task_create_local_folder(psync_syncid_t syncid, - psync_folderid_t folderid, - psync_folderid_t localfolderid); -void psync_task_delete_local_folder(psync_syncid_t syncid, - psync_folderid_t folderid, - psync_folderid_t localfolderid, - const char *remotepath); -void psync_task_delete_local_folder_recursive(psync_syncid_t syncid, - psync_folderid_t folderid, - psync_folderid_t localfolderid); -void psync_task_rename_local_folder(psync_syncid_t syncid, - psync_folderid_t folderid, - psync_folderid_t localfolderid, - psync_folderid_t newlocalparentfolderid, - const char *newname); -void psync_task_download_file(psync_syncid_t syncid, psync_fileid_t fileid, - psync_folderid_t localfolderid, const char *name); -void psync_task_download_file_silent(psync_syncid_t syncid, - psync_fileid_t fileid, - psync_folderid_t localfolderid, - const char *name); -void psync_task_rename_local_file(psync_syncid_t oldsyncid, - psync_syncid_t newsyncid, - psync_fileid_t fileid, - psync_folderid_t oldlocalfolderid, - psync_folderid_t newlocalfolderid, - const char *newname); -void psync_task_delete_local_file(psync_fileid_t fileid, - const char *remotepath); -void psync_task_delete_local_file_syncid(psync_syncid_t syncid, - psync_fileid_t fileid, - const char *remotepath); +typedef struct { + uint64_t size; + uint64_t hash; + unsigned char sha1hex[40]; +} psync_async_file_result_t; -void psync_task_create_remote_folder(psync_syncid_t syncid, - psync_folderid_t localfolderid, - const char *name); -void psync_task_upload_file(psync_syncid_t syncid, psync_fileid_t localfileid, - const char *name); -void psync_task_upload_file_silent(psync_syncid_t syncid, - psync_fileid_t localfileid, - const char *name); +typedef struct { + uint32_t error; + uint32_t errorflags; + union { + psync_async_file_result_t file; + }; +} psync_async_result_t; -/* newname should be passed here instead of reading it from localfile in time of - * renaming as there might be many pending renames and filename conflict is - * possible +typedef void (*psync_async_callback_t)(void *, psync_async_result_t *); + + +/* Important! The interface typically expect all passed pointers to be alive + * until the completion callback is called. */ -void psync_task_rename_remote_file(psync_syncid_t oldsyncid, - psync_syncid_t newsyncid, - psync_fileid_t localfileid, - psync_folderid_t newlocalparentfolderid, - const char *newname); -void psync_task_rename_remote_folder(psync_syncid_t oldsyncid, - psync_syncid_t newsyncid, - psync_fileid_t localfileid, - psync_folderid_t newlocalparentfolderid, - const char *newname); -void psync_task_delete_remote_file(psync_syncid_t syncid, - psync_fileid_t fileid); -void psync_task_delete_remote_folder(psync_syncid_t syncid, - psync_folderid_t folderid); +void ptask_stop_async(); +int ptask_download_async(psync_fileid_t fileid, const char *localpath, psync_async_callback_t cb, void *cbext); +int ptask_download_needed_async(psync_fileid_t fileid, const char *localpath, uint64_t size, const void *sha1hex, psync_async_callback_t cb, void *cbext); + +// local ops +void ptask_ldir_mk(psync_syncid_t syncid, psync_folderid_t folderid, psync_folderid_t localfolderid); +void ptask_ldir_rename(psync_syncid_t syncid, psync_folderid_t folderid, psync_folderid_t localfolderid, psync_folderid_t newlocalparentfolderid, const char *newname); +void ptask_ldir_rm(psync_syncid_t syncid, psync_folderid_t folderid, psync_folderid_t localfolderid, const char *remotepath); +void ptask_ldir_rm_r(psync_syncid_t syncid, psync_folderid_t folderid, psync_folderid_t localfolderid); +void ptask_lfile_rename(psync_syncid_t oldsyncid, psync_syncid_t newsyncid, psync_fileid_t fileid, psync_folderid_t oldlocalfolderid, psync_folderid_t newlocalfolderid, const char *newname); +void ptask_lfile_rm(psync_fileid_t fileid, const char *remotepath); +void ptask_lfile_rm_id(psync_syncid_t syncid, psync_fileid_t fileid, const char *remotepath); + +// remote ops +// +// for rename operations, "newname" should be passed here instead of reading +// it from localfile, to avoid conflict due to many pending renames +void ptask_rdir_mk(psync_syncid_t syncid, psync_folderid_t localfolderid, const char *name); +void ptask_rdir_rename(psync_syncid_t oldsyncid, psync_syncid_t newsyncid, psync_fileid_t localfileid, psync_folderid_t newlocalparentfolderid, const char *newname); +void ptask_rdir_rm(psync_syncid_t syncid, psync_folderid_t folderid); +void ptask_rfile_rename(psync_syncid_t oldsyncid, psync_syncid_t newsyncid, psync_fileid_t localfileid, psync_folderid_t newlocalparentfolderid, const char *newname); +void ptask_rfile_rm(psync_syncid_t syncid, psync_fileid_t fileid); + +// file transfer +void ptask_download(psync_syncid_t syncid, psync_fileid_t fileid, psync_folderid_t localfolderid, const char *name); +void ptask_download_q(psync_syncid_t syncid, psync_fileid_t fileid, psync_folderid_t localfolderid, const char *name); +void ptask_upload(psync_syncid_t syncid, psync_fileid_t localfileid, const char *name); +void ptask_upload_q(psync_syncid_t syncid, psync_fileid_t localfileid, const char *name); #endif diff --git a/pclsync/ptasks.c b/pclsync/ptasks.c deleted file mode 100644 index 382202f..0000000 --- a/pclsync/ptasks.c +++ /dev/null @@ -1,271 +0,0 @@ -/* - Copyright (c) 2013-2014 Anton Titov. - - Copyright (c) 2013-2014 pCloud Ltd. All rights reserved. - - Redistribution and use in source and binary forms, with or without - modification, are permitted provided that the following conditions - are met: Redistributions of source code must retain the above - copyright notice, this list of conditions and the following - disclaimer. Redistributions in binary form must reproduce the - above copyright notice, this list of conditions and the following - disclaimer in the documentation and/or other materials provided - with the distribution. Neither the name of pCloud Ltd nor the - names of its contributors may be used to endorse or promote - products derived from this software without specific prior written - permission. - - THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS - FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL pCloud - Ltd BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, - EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, - PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR - PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY - OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT - (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE - USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH - DAMAGE. -*/ - -#include -#include -#include -#include -#include -#include - -#include "pcallbacks.h" -#include "pdownload.h" -#include "plibs.h" -#include "ppathstatus.h" -#include "pstatus.h" -#include "ptasks.h" -#include "pupload.h" - -static void create_task1(unsigned long type, psync_syncid_t syncid, - uint64_t entryid, uint64_t localentryid) { - psync_sql_res *res; - res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " - "localitemid) VALUES (?, ?, ?, ?)"); - psync_sql_bind_uint(res, 1, type); - psync_sql_bind_uint(res, 2, syncid); - psync_sql_bind_uint(res, 3, entryid); - psync_sql_bind_uint(res, 4, localentryid); - psync_sql_run_free(res); -} - -static void create_task2(unsigned long type, psync_syncid_t syncid, - uint64_t entryid, uint64_t localentryid, - uint64_t newitemid, const char *name) { - psync_sql_res *res; - res = psync_sql_prep_statement( - "INSERT INTO task (type, syncid, itemid, localitemid, newitemid, name) " - "VALUES (?, ?, ?, ?, ?, ?)"); - psync_sql_bind_uint(res, 1, type); - psync_sql_bind_uint(res, 2, syncid); - psync_sql_bind_uint(res, 3, entryid); - psync_sql_bind_uint(res, 4, localentryid); - psync_sql_bind_uint(res, 5, newitemid); - psync_sql_bind_string(res, 6, name); - psync_sql_run_free(res); -} - -static void create_task3(unsigned long type, psync_syncid_t syncid, - uint64_t entryid, uint64_t localentryid, - const char *name) { - psync_sql_res *res; - res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " - "localitemid, name) VALUES (?, ?, ?, ?, ?)"); - psync_sql_bind_uint(res, 1, type); - psync_sql_bind_uint(res, 2, syncid); - psync_sql_bind_uint(res, 3, entryid); - psync_sql_bind_uint(res, 4, localentryid); - psync_sql_bind_string(res, 5, name); - psync_sql_run_free(res); -} - -static void create_task4(unsigned long type, uint64_t entryid, - const char *name) { - psync_sql_res *res; - res = psync_sql_prep_statement( - "INSERT INTO task (type, itemid, localitemid, name) VALUES (?, ?, 0, ?)"); - psync_sql_bind_uint(res, 1, type); - psync_sql_bind_uint(res, 2, entryid); - psync_sql_bind_string(res, 3, name); - psync_sql_run_free(res); -} - -static void create_task5(unsigned long type, psync_syncid_t syncid, - uint64_t entryid) { - psync_sql_res *res; - res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " - "localitemid) VALUES (?, ?, ?, 0)"); - psync_sql_bind_uint(res, 1, type); - psync_sql_bind_uint(res, 2, syncid); - psync_sql_bind_uint(res, 3, entryid); - psync_sql_run_free(res); -} - -static void create_task6(unsigned long type, psync_syncid_t syncid, - uint64_t entryid, const char *name) { - psync_sql_res *res; - res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " - "localitemid, name) VALUES (?, ?, ?, 0, ?)"); - psync_sql_bind_uint(res, 1, type); - psync_sql_bind_uint(res, 2, syncid); - psync_sql_bind_uint(res, 3, entryid); - psync_sql_bind_string(res, 4, name); - psync_sql_run_free(res); -} - -void psync_task_create_local_folder(psync_syncid_t syncid, - psync_folderid_t folderid, - psync_folderid_t localfolderid) { - create_task1(PSYNC_CREATE_LOCAL_FOLDER, syncid, folderid, localfolderid); -} - -void psync_task_delete_local_folder(psync_syncid_t syncid, - psync_folderid_t folderid, - psync_folderid_t localfolderid, - const char *remotepath) { - create_task3(PSYNC_DELETE_LOCAL_FOLDER, syncid, folderid, localfolderid, - remotepath); -} - -void psync_task_delete_local_folder_recursive(psync_syncid_t syncid, - psync_folderid_t folderid, - psync_folderid_t localfolderid) { - create_task1(PSYNC_DELREC_LOCAL_FOLDER, syncid, folderid, localfolderid); -} - -void psync_task_rename_local_folder(psync_syncid_t syncid, - psync_folderid_t folderid, - psync_folderid_t localfolderid, - psync_folderid_t newlocalparentfolderid, - const char *newname) { - create_task2(PSYNC_RENAME_LOCAL_FOLDER, syncid, folderid, localfolderid, - newlocalparentfolderid, newname); -} - -void psync_task_download_file_silent(psync_syncid_t syncid, - psync_fileid_t fileid, - psync_folderid_t localfolderid, - const char *name) { - psync_sql_res *res; - res = psync_sql_prep_statement("INSERT INTO task (type, syncid, itemid, " - "localitemid, name) VALUES (?, ?, ?, ?, ?)"); - psync_sql_bind_uint(res, 1, PSYNC_DOWNLOAD_FILE); - psync_sql_bind_uint(res, 2, syncid); - psync_sql_bind_uint(res, 3, fileid); - psync_sql_bind_uint(res, 4, localfolderid); - psync_sql_bind_string(res, 5, name); - psync_path_status_sync_folder_task_added_locked(syncid, localfolderid); - psync_sql_run_free(res); -} - -void psync_task_download_file(psync_syncid_t syncid, psync_fileid_t fileid, - psync_folderid_t localfolderid, - const char *name) { - psync_task_download_file_silent(syncid, fileid, localfolderid, name); - psync_wake_download(); - psync_status_recalc_to_download(); - psync_send_status_update(); -} - -void psync_task_rename_local_file(psync_syncid_t oldsyncid, - psync_syncid_t newsyncid, - psync_fileid_t fileid, - psync_folderid_t oldlocalfolderid, - psync_folderid_t newlocalfolderid, - const char *newname) { - psync_sql_res *res; - res = psync_sql_prep_statement( - "INSERT INTO task (type, syncid, newsyncid, itemid, localitemid, " - "newitemid, name) VALUES (?, ?, ?, ?, ?, ?, ?)"); - psync_sql_bind_uint(res, 1, PSYNC_RENAME_LOCAL_FILE); - psync_sql_bind_uint(res, 2, oldsyncid); - psync_sql_bind_uint(res, 3, newsyncid); - psync_sql_bind_uint(res, 4, fileid); - psync_sql_bind_uint(res, 5, oldlocalfolderid); - psync_sql_bind_uint(res, 6, newlocalfolderid); - psync_sql_bind_string(res, 7, newname); - psync_sql_run_free(res); -} - -void psync_task_delete_local_file(psync_fileid_t fileid, - const char *remotepath) { - create_task4(PSYNC_DELETE_LOCAL_FILE, fileid, remotepath); -} - -void psync_task_delete_local_file_syncid(psync_syncid_t syncid, - psync_fileid_t fileid, - const char *remotepath) { - create_task6(PSYNC_DELETE_LOCAL_FILE, syncid, fileid, remotepath); -} - -void psync_task_create_remote_folder(psync_syncid_t syncid, - psync_folderid_t localfolderid, - const char *name) { - create_task3(PSYNC_CREATE_REMOTE_FOLDER, syncid, 0, localfolderid, name); -} - -void psync_task_upload_file_silent(psync_syncid_t syncid, - psync_fileid_t localfileid, - const char *name) { - create_task3(PSYNC_UPLOAD_FILE, syncid, 0, localfileid, name); -} - -void psync_task_upload_file(psync_syncid_t syncid, psync_fileid_t localfileid, - const char *name) { - create_task3(PSYNC_UPLOAD_FILE, syncid, 0, localfileid, name); - psync_wake_upload(); - psync_status_recalc_to_upload_async(); -} - -void psync_task_rename_remote_file(psync_syncid_t oldsyncid, - psync_syncid_t newsyncid, - psync_fileid_t localfileid, - psync_folderid_t newlocalparentfolderid, - const char *newname) { - psync_sql_res *res; - res = psync_sql_prep_statement( - "INSERT INTO task (type, syncid, newsyncid, localitemid, newitemid, " - "name, itemid) VALUES (?, ?, ?, ?, ?, ?, 0)"); - psync_sql_bind_uint(res, 1, PSYNC_RENAME_REMOTE_FILE); - psync_sql_bind_uint(res, 2, oldsyncid); - psync_sql_bind_uint(res, 3, newsyncid); - psync_sql_bind_uint(res, 4, localfileid); - psync_sql_bind_uint(res, 5, newlocalparentfolderid); - psync_sql_bind_string(res, 6, newname); - psync_sql_run_free(res); -} - -void psync_task_rename_remote_folder(psync_syncid_t oldsyncid, - psync_syncid_t newsyncid, - psync_fileid_t localfileid, - psync_folderid_t newlocalparentfolderid, - const char *newname) { - psync_sql_res *res; - res = psync_sql_prep_statement( - "INSERT INTO task (type, syncid, newsyncid, localitemid, newitemid, " - "name, itemid) VALUES (?, ?, ?, ?, ?, ?, 0)"); - psync_sql_bind_uint(res, 1, PSYNC_RENAME_REMOTE_FOLDER); - psync_sql_bind_uint(res, 2, oldsyncid); - psync_sql_bind_uint(res, 3, newsyncid); - psync_sql_bind_uint(res, 4, localfileid); - psync_sql_bind_uint(res, 5, newlocalparentfolderid); - psync_sql_bind_string(res, 6, newname); - psync_sql_run_free(res); -} - -void psync_task_delete_remote_file(psync_syncid_t syncid, - psync_fileid_t fileid) { - create_task5(PSYNC_DELETE_REMOTE_FILE, syncid, fileid); -} - -void psync_task_delete_remote_folder(psync_syncid_t syncid, - psync_folderid_t folderid) { - create_task5(PSYNC_DELREC_REMOTE_FOLDER, syncid, folderid); -} diff --git a/pclsync/pupload.c b/pclsync/pupload.c index d441e47..8a8626e 100644 --- a/pclsync/pupload.c +++ b/pclsync/pupload.c @@ -52,7 +52,7 @@ #include "psettings.h" #include "psys.h" #include "pstatus.h" -#include "ptasks.h" +#include "ptask.h" #include "ptimer.h" #include "ptools.h" #include "pupload.h"