pcloudcc-lneely/pCloudCC/lib/pclsync/ppagecache.c

3615 lines
126 KiB
C

/* Copyright (c) 2014 Anton Titov.
* Copyright (c) 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 "ppagecache.h"
#include "psettings.h"
#include "plibs.h"
#include "ptimer.h"
#include "pnetlibs.h"
#include "pstatus.h"
#include "pcache.h"
#include "pfsupload.h"
#include "pfscrypto.h"
#include "pcrc32c.h"
#include <errno.h>
#include <string.h>
#include <stdio.h>
#define CACHE_PAGES (PSYNC_FS_MEMORY_CACHE/PSYNC_FS_PAGE_SIZE)
#define CACHE_HASH (CACHE_PAGES/2)
#define PAGE_WAITER_HASH 1024
#define DB_CACHE_UPDATE_HASH (32*1024)
#define PAGE_TYPE_FREE 0
#define PAGE_TYPE_READ 1
#define PAGE_TYPE_CACHE 2
#define PAGE_TASK_TYPE_CREAT 0
#define PAGE_TASK_TYPE_MODIFY 1
#define pagehash_by_hash_and_pageid(hash, pageid) (((hash)+(pageid))%CACHE_HASH)
#define waiterhash_by_hash_and_pageid(hash, pageid) (((hash)+(pageid))%PAGE_WAITER_HASH)
#define lock_wait(hash) pthread_mutex_lock(&wait_page_mutex)
#define unlock_wait(hash) pthread_mutex_unlock(&wait_page_mutex)
typedef struct {
psync_list list;
psync_list flushlist;
char *page;
uint64_t hash;
uint64_t pageid;
time_t lastuse;
uint32_t size;
uint32_t usecnt;
uint32_t flushpageid;
uint32_t crc;
uint8_t type;
} psync_cache_page_t;
typedef struct {
uint64_t pagecacheid;
time_t lastuse;
uint32_t usecnt;
} psync_cachepage_to_update;
typedef struct {
/* list is an element of hash table for pages */
psync_list list;
/* list is root of a listpage elements of psync_page_waiter_t, if empty, nobody waits for this page */
psync_list waiters;
uint64_t hash;
uint64_t pageid;
psync_fileid_t fileid;
} psync_page_wait_t;
typedef struct {
/* listpage is node element of psync_page_wait_t waiters list */
psync_list listpage;
/* listwaiter is node element of pages that are needed for current request */
psync_list listwaiter;
pthread_cond_t cond;
psync_page_wait_t *waiting_for;
char *buff;
uint32_t pageidx;
uint32_t rsize;
uint32_t size;
uint32_t off;
int error;
uint8_t ready;
} psync_page_waiter_t;
typedef struct {
psync_list list;
uint64_t offset;
uint64_t length;
} psync_request_range_t;
typedef struct {
psync_list ranges;
psync_openfile_t *of;
psync_fileid_t fileid;
uint64_t hash;
int needkey;
} psync_request_t;
typedef struct {
uint64_t hash;
psync_tree tree;
binresult *urls;
uint32_t refcnt;
uint32_t status;
} psync_urls_t;
typedef struct _psync_crypto_auth_page {
psync_list list;
psync_page_waiter_t *waiter;
struct _psync_crypto_auth_page *parent;
uint64_t firstpageid;
uint32_t size;
uint32_t idinparent;
uint32_t level;
psync_crypto_auth_sector_t auth;
} psync_crypto_auth_page;
typedef struct {
psync_crypto_auth_page *authpage;
psync_page_waiter_t *waiter;
char *buff;
uint32_t pagesize;
uint8_t freebuff;
} psync_crypto_data_page;
static psync_list cache_hash[CACHE_HASH];
static uint32_t cache_pages_in_hash=0;
static uint32_t cache_pages_free;
static int cache_pages_reset=1;
static psync_list free_pages;
static psync_list wait_page_hash[PAGE_WAITER_HASH];
static char *pages_base;
static uint32_t free_page_waiters=0;
static int flush_page_running=0;
static psync_cachepage_to_update cachepages_to_update[DB_CACHE_UPDATE_HASH];
static uint32_t cachepages_to_update_cnt=0;
static uint32_t free_db_pages;
static pthread_mutex_t clean_cache_mutex=PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t clean_cache_cond=PTHREAD_COND_INITIALIZER;
static pthread_mutex_t cache_mutex=PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t flush_cache_mutex=PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t free_page_cond=PTHREAD_COND_INITIALIZER;
static pthread_mutex_t url_cache_mutex=PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t url_cache_cond=PTHREAD_COND_INITIALIZER;
static pthread_mutex_t wait_page_mutex;
static pthread_cond_t enc_key_cond=PTHREAD_COND_INITIALIZER;
static uint32_t clean_cache_stoppers=0;
static uint32_t clean_cache_waiters=0;
static uint32_t clean_cache_in_progress=0;
static int flushedbetweentimers=0;
static int flushcacherun=0;
static int upload_to_cache_thread_run=0;
static uint64_t db_cache_in_pages;
static uint64_t db_cache_max_page;
static psync_file_t readcache=INVALID_HANDLE_VALUE;
static psync_tree *url_cache_tree=PSYNC_TREE_EMPTY;
static int flush_pages(int nosleep);
static void flush_pages_noret(){
flush_pages(0);
}
static psync_cache_page_t *psync_pagecache_get_free_page_if_available(){
psync_cache_page_t *page;
int runthread;
runthread=0;
pthread_mutex_lock(&cache_mutex);
if (unlikely(cache_pages_free<=CACHE_PAGES*25/100 && !flushcacherun)){
flushcacherun=1;
runthread=1;
}
if (likely(!psync_list_isempty(&free_pages)))
page=psync_list_remove_head_element(&free_pages, psync_cache_page_t, list);
else
page=NULL;
pthread_mutex_unlock(&cache_mutex);
if (runthread)
psync_run_thread("flush pages get free page ifav", flush_pages_noret);
return page;
}
static psync_cache_page_t *psync_pagecache_get_free_page(int runflushcacheinside){
psync_cache_page_t *page;
int runthread;
runthread=0;
pthread_mutex_lock(&cache_mutex);
if (unlikely(cache_pages_free<=CACHE_PAGES*25/100 && !flushcacherun)){
flushcacherun=1;
if (runflushcacheinside){
pthread_mutex_unlock(&cache_mutex);
debug(D_NOTICE, "running flush cache on this thread");
flush_pages(2);
pthread_mutex_lock(&cache_mutex);
}
else
runthread=1;
}
if (likely(!psync_list_isempty(&free_pages)))
page=psync_list_remove_head_element(&free_pages, psync_cache_page_t, list);
else{
if (flush_page_running){
debug(D_NOTICE, "no free pages, but somebody is flushing cache, waiting for a page");
do {
free_page_waiters++;
pthread_cond_wait(&free_page_cond, &cache_mutex);
free_page_waiters--;
} while (flush_page_running && psync_list_isempty(&free_pages));
}
if (psync_list_isempty(&free_pages)){
debug(D_NOTICE, "no free pages, flushing cache");
pthread_mutex_unlock(&cache_mutex);
flush_pages(1);
pthread_mutex_lock(&cache_mutex);
while (unlikely(psync_list_isempty(&free_pages))){
pthread_mutex_unlock(&cache_mutex);
debug(D_NOTICE, "no free pages after flush, sleeping");
psync_milisleep(200);
flush_pages(1);
pthread_mutex_lock(&cache_mutex);
}
}
else
debug(D_NOTICE, "waited for a free page");
page=psync_list_remove_head_element(&free_pages, psync_cache_page_t, list);
}
cache_pages_free--;
pthread_mutex_unlock(&cache_mutex);
if (runthread)
psync_run_thread("flush pages get free page", flush_pages_noret);
return page;
}
static int psync_api_send_read_request(psync_socket *api, psync_fileid_t fileid, uint64_t hash, uint64_t offset, uint64_t length){
binparam params[]={P_STR("auth", psync_my_auth), P_NUM("fileid", fileid), P_NUM("hash", hash), P_NUM("offset", offset), P_NUM("count", length)};
return send_command_no_res(api, "readfile", params)==PTR_OK?0:-1;
}
static int psync_api_send_read_request_thread(psync_socket *api, psync_fileid_t fileid, uint64_t hash, uint64_t offset, uint64_t length){
binparam params[]={P_STR("auth", psync_my_auth), P_NUM("fileid", fileid), P_NUM("hash", hash), P_NUM("offset", offset), P_NUM("count", length)};
return send_command_no_res_thread(api, "readfile", params)==PTR_OK?0:-1;
}
static void psync_pagecache_send_page_wait_page(psync_page_wait_t *pw, psync_cache_page_t *page){
psync_page_waiter_t *pwt;
psync_list_del(&pw->list);
psync_list_for_each_element(pwt, &pw->waiters, psync_page_waiter_t, listpage){
page->usecnt++;
if (pwt->off+pwt->size>page->size){
if (pwt->off>=page->size)
pwt->rsize=0;
else
pwt->rsize=page->size-pwt->off;
}
else
pwt->rsize=pwt->size;
memcpy(pwt->buff, page->page+pwt->off, pwt->rsize);
pwt->error=0;
pwt->ready=1;
pwt->waiting_for=NULL;
pthread_cond_broadcast(&pwt->cond);
}
psync_free(pw);
}
static void psync_pagecache_return_free_page_locked(psync_cache_page_t *page){
psync_list_add_head(&free_pages, &page->list);
cache_pages_free++;
}
static void psync_pagecache_return_free_page(psync_cache_page_t *page){
pthread_mutex_lock(&cache_mutex);
psync_pagecache_return_free_page_locked(page);
pthread_mutex_unlock(&cache_mutex);
}
static int psync_pagecache_read_range_from_api(psync_request_t *request, psync_request_range_t *range, psync_socket *api){
uint64_t first_page_id, dlen;
psync_page_wait_t *pw;
psync_cache_page_t *page;
binresult *res;
psync_uint_t len, i, h;
int rb;
first_page_id=range->offset/PSYNC_FS_PAGE_SIZE;
len=range->length/PSYNC_FS_PAGE_SIZE;
res=get_result_thread(api);
if (unlikely_log(!res))
return -2;
dlen=psync_find_result(res, "result", PARAM_NUM)->num;
if (unlikely(dlen)){
psync_free(res);
debug(D_WARNING, "readfile returned error %lu", (long unsigned)dlen);
psync_process_api_error(dlen);
return -2;
}
dlen=psync_find_result(res, "data", PARAM_DATA)->num;
psync_free(res);
for (i=0; i<len; i++){
page=psync_pagecache_get_free_page(0);
rb=psync_socket_readall_download_thread(api, page->page, dlen<PSYNC_FS_PAGE_SIZE?dlen:PSYNC_FS_PAGE_SIZE);
if (unlikely_log(rb<=0)){
psync_pagecache_return_free_page(page);
psync_timer_notify_exception();
return i==0?-2:-1;
}
dlen-=rb;
page->hash=request->hash;
page->pageid=first_page_id+i;
page->lastuse=psync_timer_time();
page->size=rb;
page->usecnt=0;
page->crc=psync_crc32c(PSYNC_CRC_INITIAL, page->page, rb);
page->type=PAGE_TYPE_READ;
h=waiterhash_by_hash_and_pageid(page->hash, page->pageid);
lock_wait(page->hash);
psync_list_for_each_element(pw, &wait_page_hash[h], psync_page_wait_t, list)
if (pw->hash==page->hash && pw->pageid==page->pageid){
psync_pagecache_send_page_wait_page(pw, page);
break;
}
unlock_wait(page->hash);
pthread_mutex_lock(&cache_mutex);
psync_list_add_tail(&cache_hash[pagehash_by_hash_and_pageid(page->hash, page->pageid)], &page->list);
cache_pages_in_hash++;
pthread_mutex_unlock(&cache_mutex);
}
return 0;
}
typedef struct {
psync_list list;
pthread_cond_t cond;
psync_socket *api;
} shared_api_waiter_t;
static pthread_mutex_t sharedapi_mutex=PTHREAD_MUTEX_INITIALIZER;
static psync_socket *sharedapi=NULL;
static psync_list sharedapiwaiters=PSYNC_LIST_STATIC_INIT(sharedapiwaiters);
static void mark_api_shared(psync_socket *api){
pthread_mutex_lock(&sharedapi_mutex);
if (!sharedapi)
sharedapi=api;
pthread_mutex_unlock(&sharedapi_mutex);
}
static void signal_all_waiters(){
shared_api_waiter_t *waiter;
while (!psync_list_isempty(&sharedapiwaiters)){
waiter=psync_list_remove_head_element(&sharedapiwaiters, shared_api_waiter_t, list);
waiter->api=(psync_socket *)-1;
pthread_cond_signal(&waiter->cond);
}
}
static void mark_shared_api_bad(psync_socket *api){
pthread_mutex_lock(&sharedapi_mutex);
if (sharedapi==api){
sharedapi=NULL;
signal_all_waiters();
}
pthread_mutex_unlock(&sharedapi_mutex);
}
static int pass_shared_api(psync_socket *api){
shared_api_waiter_t *waiter;
int ret;
pthread_mutex_lock(&sharedapi_mutex);
if (api!=sharedapi)
ret=-1;
else if (psync_list_isempty(&sharedapiwaiters)){
ret=-1;
sharedapi=NULL;
}
else{
ret=0;
waiter=psync_list_remove_head_element(&sharedapiwaiters, shared_api_waiter_t, list);
waiter->api=api;
pthread_cond_signal(&waiter->cond);
debug(D_NOTICE, "passing shared api connection");
}
pthread_mutex_unlock(&sharedapi_mutex);
return ret;
}
static psync_socket *get_shared_api(){
pthread_mutex_lock(&sharedapi_mutex);
if (sharedapi)
return sharedapi; // not supposed to unlock, it will happen in wait_shared_api
pthread_mutex_unlock(&sharedapi_mutex);
return NULL;
}
static void release_bad_shared_api(psync_socket *api){
if (sharedapi==api){
sharedapi=NULL;
signal_all_waiters();
}
pthread_mutex_unlock(&sharedapi_mutex);
}
static int wait_shared_api(){
shared_api_waiter_t *waiter;
psync_socket *capi;
int ret;
capi=sharedapi;
waiter=psync_new(shared_api_waiter_t);
pthread_cond_init(&waiter->cond, NULL);
waiter->api=NULL;
psync_list_add_tail(&sharedapiwaiters, &waiter->list);
debug(D_NOTICE, "waiting for shared API connection");
do {
pthread_cond_wait(&waiter->cond, &sharedapi_mutex);
} while (!waiter->api);
if (waiter->api!=capi){
assertw(waiter->api==(psync_socket *)-1);
ret=-1;
}
else{
debug(D_NOTICE, "waited for shared API connection");
ret=0;
}
pthread_mutex_unlock(&sharedapi_mutex);
pthread_cond_destroy(&waiter->cond);
psync_free(waiter);
return ret;
}
static void set_urls(psync_urls_t *urls, binresult *res){
pthread_mutex_lock(&url_cache_mutex);
if (res){
urls->status=1;
urls->urls=res;
if (urls->refcnt++>0)
pthread_cond_broadcast(&url_cache_cond);
}
else{
psync_tree_del(&url_cache_tree, &urls->tree);
if (urls->refcnt){
urls->status=2;
pthread_cond_broadcast(&url_cache_cond);
}
else
psync_free(urls);
}
pthread_mutex_unlock(&url_cache_mutex);
}
static void psync_pagecache_set_bad_encoder(psync_openfile_t *of){
psync_fs_lock_file(of);
if (likely(of->encoder==PSYNC_CRYPTO_LOADING_SECTOR_ENCODER)){
of->encoder=PSYNC_CRYPTO_FAILED_SECTOR_ENCODER;
pthread_cond_broadcast(&enc_key_cond);
}
pthread_mutex_unlock(&of->mutex);
}
static int send_key_request(psync_socket *api, psync_request_t *request){
binparam params[]={P_STR("auth", psync_my_auth), P_NUM("fileid", request->fileid)};
return send_command_no_res(api, "crypto_getfilekey", params)!=PTR_OK?-1:0;
}
static int get_urls(psync_request_t *request, psync_urls_t *urls){
binparam params[]={P_STR("auth", psync_my_auth), P_NUM("fileid", request->fileid), P_NUM("hash", request->hash),
P_STR("timeformat", "timestamp"), P_BOOL("skipfilename", 1)};
psync_socket *api;
binresult *ret;
psync_request_range_t *range;
psync_list *l1, *l2;
const binresult *hosts;
uint64_t totalreqlen;
unsigned long result;
int tries;
debug(D_NOTICE, "getting file URLs of fileid %lu, hash %lu together with requests%s",
(unsigned long)request->fileid, (unsigned long)request->hash, request->needkey?" and encryption key":"");
tries=0;
while (tries++<=5){
api=psync_apipool_get();
if (unlikely_log(!api))
continue;
psync_socket_set_write_buffered(api);
if (unlikely(send_command_no_res(api, "getfilelink", params)!=PTR_OK))
goto err1;
if (request->needkey && send_key_request(api, request))
goto err1;
totalreqlen=0;
psync_list_for_each_element(range, &request->ranges, psync_request_range_t, list){
debug(D_NOTICE, "sending request for offset %lu, size %lu to API", (unsigned long)range->offset, (unsigned long)range->length);
if (unlikely(psync_api_send_read_request(api, request->fileid, request->hash, range->offset, range->length)))
goto err1;
totalreqlen+=range->length;
}
mark_api_shared(api);
ret=get_result_thread(api);
if (unlikely_log(!ret)){
mark_shared_api_bad(api);
goto err1;
}
result=psync_find_result(ret, "result", PARAM_NUM)->num;
if (unlikely(result!=0)){
debug(D_WARNING, "getfilelink returned error %lu", result);
psync_free(ret);
mark_shared_api_bad(api);
psync_apipool_release_bad(api);
psync_process_api_error(result);
break;
}
hosts=psync_find_result(ret, "hosts", PARAM_ARRAY);
debug(D_NOTICE, "got file URLs of fileid %lu, hash %lu", (unsigned long)request->fileid, (unsigned long)request->hash);
if (likely_log(hosts->length && hosts->array[0]->type==PARAM_STR) && request->of->initialsize>totalreqlen)
psync_http_connect_and_cache_host(hosts->array[0]->str);
/*if (of->initialsize>=PSYNC_FS_FILESIZE_FOR_2CONN && hosts->length>1 && hosts->array[1]->type==PARAM_STR)
psync_http_connect_and_cache_host(hosts->array[1]->str);*/
set_urls(urls, ret);
if (request->needkey){
psync_crypto_aes256_sector_encoder_decoder_t enc;
ret=get_result_thread(api);
if (unlikely_log(!ret))
goto err3;
result=psync_find_result(ret, "result", PARAM_NUM)->num;
if (unlikely(result!=0)){
debug(D_WARNING, "crypto_getfilekey returned error %lu", result);
psync_process_api_error(result);
goto err4;
}
enc=psync_cloud_crypto_get_file_encoder_from_binresult(request->fileid, ret);
if (unlikely_log(psync_crypto_is_error(enc)))
goto err4;
debug(D_NOTICE, "got key for fileid %lu", (unsigned long)request->fileid);
psync_free(ret);
psync_fs_lock_file(request->of);
if (likely_log(request->of->encoder==PSYNC_CRYPTO_LOADING_SECTOR_ENCODER)){
request->of->encoder=enc;
pthread_cond_broadcast(&enc_key_cond);
}
else
psync_cloud_crypto_release_file_encoder(request->fileid, request->hash, enc);
pthread_mutex_unlock(&request->of->mutex);
request->needkey=0;
}
psync_list_for_each_safe(l1, l2, &request->ranges){
range=psync_list_element(l1, psync_request_range_t, list);
if (psync_pagecache_read_range_from_api(request, range, api))
goto err2;
psync_list_del(l1);
debug(D_NOTICE, "request for offset %lu, size %lu read from API", (unsigned long)range->offset, (unsigned long)range->length);
psync_free(range);
}
if (pass_shared_api(api))
psync_apipool_release(api);
return 0;
err1:
psync_apipool_release_bad(api);
}
return -1;
err4:
psync_free(ret);
err3:
if (request->needkey)
psync_pagecache_set_bad_encoder(request->of);
err2:
mark_shared_api_bad(api);
psync_apipool_release_bad(api);
return 0;
}
static psync_urls_t *get_urls_for_request(psync_request_t *req){
char buff[16];
psync_tree *el, **pel;
psync_urls_t *urls;
binresult *res;
int64_t d;
pthread_mutex_lock(&url_cache_mutex);
el=url_cache_tree;
pel=&url_cache_tree;
d=-1;
while (el){
urls=psync_tree_element(el, psync_urls_t, tree);
d=req->hash-urls->hash;
if (d==0)
break;
else if (d<0){
if (el->left)
el=el->left;
else{
pel=&el->left;
break;
}
}
else{
if (el->right)
el=el->right;
else{
pel=&el->right;
break;
}
}
}
if (d==0){
urls->refcnt++;
while (urls->status==0)
pthread_cond_wait(&url_cache_cond, &url_cache_mutex);
if (likely(urls->status==1)){
pthread_mutex_unlock(&url_cache_mutex);
return urls;
}
if (--urls->refcnt==0)
psync_free(urls);
pthread_mutex_unlock(&url_cache_mutex);
return NULL;
}
urls=psync_new(psync_urls_t);
urls->hash=req->hash;
urls->refcnt=0;
urls->status=0;
*pel=&urls->tree;
psync_tree_added_at(&url_cache_tree, el, &urls->tree);
pthread_mutex_unlock(&url_cache_mutex);
psync_get_string_id(buff, "URLS", req->hash);
res=(binresult *)psync_cache_get(buff);
if (res){
set_urls(urls, res);
return urls;
}
if (get_urls(req, urls)){
set_urls(urls, NULL);
return NULL;
}
else
return urls;
}
static void release_urls(psync_urls_t *urls){
pthread_mutex_lock(&url_cache_mutex);
if (--urls->refcnt==0){
if (likely(urls->status==1)){
char buff[16];
time_t ctime, etime;
psync_tree_del(&url_cache_tree, &urls->tree);
ctime=psync_timer_time();
etime=psync_find_result(urls->urls, "expires", PARAM_NUM)->num;
if (etime>ctime+3600){
psync_get_string_id(buff, "URLS", urls->hash);
psync_cache_add(buff, urls->urls, etime-ctime-3600, psync_free, 2);
urls->urls=NULL;
}
}
pthread_mutex_unlock(&url_cache_mutex);
psync_free(urls->urls);
psync_free(urls);
return;
}
pthread_mutex_unlock(&url_cache_mutex);
}
static void release_bad_urls(psync_urls_t *urls){
pthread_mutex_lock(&url_cache_mutex);
if (urls->status==1){
urls->status=2;
psync_tree_del(&url_cache_tree, &urls->tree);
}
if (--urls->refcnt)
urls=NULL;
pthread_mutex_unlock(&url_cache_mutex);
if (urls){
psync_free(urls->urls);
psync_free(urls);
}
}
static uint64_t offset_round_down_to_page(uint64_t offset){
return offset&~(((uint64_t)PSYNC_FS_PAGE_SIZE)-1);
}
static uint64_t size_round_up_to_page(uint64_t size){
return ((size-1)|(((uint64_t)PSYNC_FS_PAGE_SIZE)-1))+1;
}
static int has_page_in_cache_by_hash(uint64_t hash, uint64_t pageid){
psync_cache_page_t *page;
psync_uint_t h;
h=pagehash_by_hash_and_pageid(hash, pageid);
pthread_mutex_lock(&cache_mutex);
psync_list_for_each_element(page, &cache_hash[h], psync_cache_page_t, list)
if (page->hash==hash && page->pageid==pageid){
pthread_mutex_unlock(&cache_mutex);
return 1;
}
pthread_mutex_unlock(&cache_mutex);
return 0;
}
static unsigned char *has_pages_in_db(uint64_t hash, uint64_t pageid, uint32_t pagecnt, int readahead){
psync_sql_res *res;
psync_uint_row row;
unsigned char *ret;
uint64_t fromid;
uint32_t fcnt;
if (unlikely(!pagecnt))
return NULL;
ret=psync_new_cnt(unsigned char, pagecnt);
memset(ret, 0, pagecnt);
fromid=0;
fcnt=0;
res=psync_sql_query_rdlock("SELECT pageid, id FROM pagecache WHERE type=+"NTO_STR(PAGE_TYPE_READ)" AND hash=? AND pageid>=? AND pageid<? ORDER BY pageid");
psync_sql_bind_uint(res, 1, hash);
psync_sql_bind_uint(res, 2, pageid);
psync_sql_bind_uint(res, 3, pageid+pagecnt);
while ((row=psync_sql_fetch_rowint(res))){
ret[row[0]-pageid]=1;
if (row[1]==fromid+fcnt)
fcnt++;
else{
if (fcnt && readahead)
psync_file_readahead(readcache, fromid*PSYNC_FS_PAGE_SIZE, fcnt*PSYNC_FS_PAGE_SIZE);
fromid=row[1];
fcnt=1;
}
}
psync_sql_free_result(res);
if (fcnt && readahead)
psync_file_readahead(readcache, fromid*PSYNC_FS_PAGE_SIZE, fcnt*PSYNC_FS_PAGE_SIZE);
return ret;
}
static int has_page_in_db(uint64_t hash, uint64_t pageid){
psync_sql_res *res;
psync_uint_row row;
res=psync_sql_query_rdlock("SELECT pageid FROM pagecache WHERE type=+"NTO_STR(PAGE_TYPE_READ)" AND hash=? AND pageid=?");
psync_sql_bind_uint(res, 1, hash);
psync_sql_bind_uint(res, 2, pageid);
row=psync_sql_fetch_rowint(res);
psync_sql_free_result(res);
return row!=NULL;
}
static psync_int_t check_page_in_memory_by_hash(uint64_t hash, uint64_t pageid, char *buff, psync_uint_t size, psync_uint_t off){
psync_cache_page_t *page;
psync_uint_t h;
psync_int_t ret;
uint32_t crc;
time_t tm;
ret=-1;
h=pagehash_by_hash_and_pageid(hash, pageid);
pthread_mutex_lock(&cache_mutex);
psync_list_for_each_element(page, &cache_hash[h], psync_cache_page_t, list)
if (page->hash==hash && page->pageid==pageid){
psync_prefetch(page->page);
tm=psync_timer_time();
if (tm>page->lastuse+5){
page->usecnt++;
page->lastuse=tm;
}
crc=psync_crc32c(PSYNC_CRC_INITIAL, page->page, page->size);
if (unlikely(crc!=page->crc)){
debug(D_WARNING, "memory page CRC does not match %u!=%u, this is most likely memory fault or corruption, pageid %u",
(unsigned)crc, (unsigned)page->crc, (unsigned)page->pageid);
psync_list_del(&page->list);
psync_pagecache_return_free_page_locked(page);
cache_pages_in_hash--;
break;
}
if (size+off>page->size){
if (off>page->size)
size=0;
else
size=page->size-off;
}
memcpy(buff, page->page+off, size);
ret=size;
}
pthread_mutex_unlock(&cache_mutex);
return ret;
}
static int switch_memory_page_to_hash(uint64_t oldhash, uint64_t newhash, uint64_t pageid){
psync_cache_page_t *page;
psync_uint_t ho, hn;
ho=pagehash_by_hash_and_pageid(oldhash, pageid);
hn=pagehash_by_hash_and_pageid(newhash, pageid);
pthread_mutex_lock(&cache_mutex);
psync_list_for_each_element(page, &cache_hash[ho], psync_cache_page_t, list)
if (page->hash==oldhash && page->pageid==pageid && page->type==PAGE_TYPE_READ){
psync_list_del(&page->list);
page->hash=newhash;
psync_list_add_tail(&cache_hash[hn], &page->list);
pthread_mutex_unlock(&cache_mutex);
return 1;
}
pthread_mutex_unlock(&cache_mutex);
return 0;
}
typedef struct {
uint32_t lastuse;
uint32_t id;
uint16_t usecnt;
int8_t isfirst;
int8_t isxfirst;
} pagecache_entry;
static int pagecache_entry_cmp_lastuse(const void *p1, const void *p2){
const pagecache_entry *e1, *e2;
e1=(const pagecache_entry *)p1;
e2=(const pagecache_entry *)p2;
return (int)((int64_t)e2->lastuse-(int64_t)e1->lastuse);
}
static int pagecache_entry_cmp_usecnt_lastuse2(const void *p1, const void *p2){
const pagecache_entry *e1, *e2;
e1=(const pagecache_entry *)p1;
e2=(const pagecache_entry *)p2;
if (e1->usecnt>=2 && e2->usecnt<2)
return -1;
else if (e2->usecnt>=2 && e1->usecnt<2)
return 1;
else
return (int)((int64_t)e2->lastuse-(int64_t)e1->lastuse);
}
static int pagecache_entry_cmp_usecnt_lastuse4(const void *p1, const void *p2){
const pagecache_entry *e1, *e2;
e1=(const pagecache_entry *)p1;
e2=(const pagecache_entry *)p2;
if (e1->usecnt>=4 && e2->usecnt<4)
return -1;
else if (e2->usecnt>=4 && e1->usecnt<4)
return 1;
else
return (int)((int64_t)e2->lastuse-(int64_t)e1->lastuse);
}
static int pagecache_entry_cmp_usecnt_lastuse8(const void *p1, const void *p2){
const pagecache_entry *e1, *e2;
e1=(const pagecache_entry *)p1;
e2=(const pagecache_entry *)p2;
if (e1->usecnt>=8 && e2->usecnt<8)
return -1;
else if (e2->usecnt>=8 && e1->usecnt<8)
return 1;
else
return (int)((int64_t)e2->lastuse-(int64_t)e1->lastuse);
}
static int pagecache_entry_cmp_usecnt_lastuse16(const void *p1, const void *p2){
const pagecache_entry *e1, *e2;
e1=(const pagecache_entry *)p1;
e2=(const pagecache_entry *)p2;
if (e1->usecnt>=16 && e2->usecnt<16)
return -1;
else if (e2->usecnt>=16 && e1->usecnt<16)
return 1;
else
return (int)((int64_t)e2->lastuse-(int64_t)e1->lastuse);
}
static int pagecache_entry_cmp_id(const void *p1, const void *p2){
return (int)((const pagecache_entry *)p1)->id-(int)((const pagecache_entry *)p2)->id;
}
static int pagecache_entry_cmp_first_pages(const void *p1, const void *p2){
const pagecache_entry *e1, *e2;
int d;
e1=(const pagecache_entry *)p1;
e2=(const pagecache_entry *)p2;
d=(int)e2->isfirst-(int)e1->isfirst;
if (d)
return d;
else if (e1->isfirst)
return (int)((int64_t)e2->lastuse-(int64_t)e1->lastuse);
else{
d=(int)e2->isxfirst-(int)e1->isxfirst;
if (d)
return d;
else
return (int)((int64_t)e2->lastuse-(int64_t)e1->lastuse);
}
}
static int pagecache_entry_cmp_xfirst_pages(const void *p1, const void *p2){
const pagecache_entry *e1, *e2;
int d;
e1=(const pagecache_entry *)p1;
e2=(const pagecache_entry *)p2;
d=(int)e2->isxfirst-(int)e1->isxfirst;
if (d)
return d;
else
return (int)((int64_t)e2->lastuse-(int64_t)e1->lastuse);
}
/* sum should be around 90-95 percent, so after a run cache get smaller */
#define PSYNC_FS_CACHE_LRU_PERCENT 40
#define PSYNC_FS_CACHE_LRU2_PERCENT 20
#define PSYNC_FS_CACHE_LRU4_PERCENT 15
#define PSYNC_FS_CACHE_LRU8_PERCENT 10
#define PSYNC_FS_CACHE_LRU16_PERCENT 5
/* first pages percent pages are first reserved for the least recently used first pages and the above percents are
* only applied to the remainder, so it is OK if first pages percent plus all the above go above 100%
*/
#define PSYNC_FS_CACHE_LRU_FIRST_PAGES_PERCENT 15
#define PSYNC_FS_CACHE_LRU_XFIRST_PAGES_PERCENT 5
#define PSYNC_FS_FIRST_PAGES_UNDER_ID (PSYNC_FS_MIN_READAHEAD_START/PSYNC_FS_PAGE_SIZE)
#define PSYNC_FS_XFIRST_PAGES_UNDER_ID (1024*1024/PSYNC_FS_PAGE_SIZE)
static void clean_cache(){
psync_sql_res *res;
uint64_t ocnt, cnt, rcnt, i, e;
psync_uint_row row;
pagecache_entry *entries, *oentries;
debug(D_NOTICE, "cleaning cache, free cache pages %u", (unsigned)free_db_pages);
if (pthread_mutex_trylock(&clean_cache_mutex)){
debug(D_NOTICE, "cache clean already in progress, skipping");
return;
}
while (clean_cache_stoppers){
clean_cache_waiters++;
pthread_cond_wait(&clean_cache_cond, &clean_cache_mutex);
if (--clean_cache_waiters){
// leave the last waiter to do the job
pthread_mutex_unlock(&clean_cache_mutex);
return;
}
}
cnt=psync_sql_cellint("SELECT MAX(id) FROM pagecache", 0);
if (!cnt){
pthread_mutex_unlock(&clean_cache_mutex);
debug(D_NOTICE, "no entries in pagecache, cancelling cache clean");
return;
}
clean_cache_in_progress=1;
psync_sql_sync();
entries=(pagecache_entry *)psync_malloc(cnt*sizeof(pagecache_entry));
i=0;
e=0;
while (i<cnt){
res=psync_sql_query_rdlock("SELECT id, pageid, lastuse, usecnt, type FROM pagecache WHERE id>? ORDER BY id LIMIT 50000");
psync_sql_bind_uint(res, 1, e);
row=psync_sql_fetch_rowint(res);
if (unlikely(!row)){
psync_sql_free_result(res);
break;
}
do{
if (unlikely(i>=cnt))
break;
e=row[0];
if (likely(row[4]==PAGE_TYPE_READ)){
entries[i].lastuse=row[2];
entries[i].id=row[0];
if (row[3]>UINT16_MAX)
entries[i].usecnt=UINT16_MAX;
else
entries[i].usecnt=row[3];
entries[i].isfirst=row[2]<PSYNC_FS_FIRST_PAGES_UNDER_ID;
entries[i].isxfirst=row[2]<PSYNC_FS_XFIRST_PAGES_UNDER_ID;
i++;
if ((i&0x3ff)==0x3ff && psync_sql_has_waiters())
break;
}
row=psync_sql_fetch_rowint(res);
} while (row);
psync_sql_free_result(res);
if (free_db_pages)
psync_milisleep(1);
}
ocnt=cnt=i;
oentries=entries;
debug(D_NOTICE, "read %lu entries", (unsigned long)cnt);
rcnt=PSYNC_FS_CACHE_LRU_FIRST_PAGES_PERCENT*ocnt/100;
psync_qpartition(entries, cnt, rcnt, sizeof(pagecache_entry), pagecache_entry_cmp_first_pages);
cnt-=rcnt;
entries+=rcnt;
debug(D_NOTICE, "sorted first pages, reserved %lu pages, continuing with %lu entries", (unsigned long)rcnt, (unsigned long)cnt);
rcnt=PSYNC_FS_CACHE_LRU_XFIRST_PAGES_PERCENT*ocnt/100;
psync_qpartition(entries, cnt, rcnt, sizeof(pagecache_entry), pagecache_entry_cmp_xfirst_pages);
cnt-=rcnt;
entries+=rcnt;
debug(D_NOTICE, "sorted extended first pages, reserved %lu pages, continuing with %lu entries", (unsigned long)rcnt, (unsigned long)cnt);
ocnt=cnt;
rcnt=PSYNC_FS_CACHE_LRU_PERCENT*ocnt/100;
psync_qpartition(entries, cnt, rcnt, sizeof(pagecache_entry), pagecache_entry_cmp_lastuse);
cnt-=rcnt;
entries+=rcnt;
debug(D_NOTICE, "sorted entries by lastuse, reserved %lu pages, continuing with %lu oldest entries", (unsigned long)rcnt, (unsigned long)cnt);
rcnt=PSYNC_FS_CACHE_LRU2_PERCENT*ocnt/100;
psync_qpartition(entries, cnt, rcnt, sizeof(pagecache_entry), pagecache_entry_cmp_usecnt_lastuse2);
cnt-=rcnt;
entries+=rcnt;
debug(D_NOTICE, "sorted entries by more than 2 uses and lastuse, reserved %lu pages, continuing with %lu entries", (unsigned long)rcnt, (unsigned long)cnt);
rcnt=PSYNC_FS_CACHE_LRU4_PERCENT*ocnt/100;
psync_qpartition(entries, cnt, rcnt, sizeof(pagecache_entry), pagecache_entry_cmp_usecnt_lastuse4);
cnt-=rcnt;
entries+=rcnt;
debug(D_NOTICE, "sorted entries by more than 4 uses and lastuse, reserved %lu pages, continuing with %lu entries", (unsigned long)rcnt, (unsigned long)cnt);
rcnt=PSYNC_FS_CACHE_LRU8_PERCENT*ocnt/100;
psync_qpartition(entries, cnt, rcnt, sizeof(pagecache_entry), pagecache_entry_cmp_usecnt_lastuse8);
cnt-=rcnt;
entries+=rcnt;
debug(D_NOTICE, "sorted entries by more than 8 uses and lastuse, reserved %lu pages, continuing with %lu entries", (unsigned long)rcnt, (unsigned long)cnt);
rcnt=PSYNC_FS_CACHE_LRU16_PERCENT*ocnt/100;
psync_qpartition(entries, cnt, rcnt, sizeof(pagecache_entry), pagecache_entry_cmp_usecnt_lastuse16);
cnt-=rcnt;
entries+=rcnt;
debug(D_NOTICE, "sorted entries by more than 16 uses and lastuse, reserved %lu pages, deleting %lu entries", (unsigned long)rcnt, (unsigned long)cnt);
psync_pqsort(entries, cnt, cnt, sizeof(pagecache_entry), pagecache_entry_cmp_id);
debug(D_NOTICE, "sorted entries to delete by id to help the SQL");
psync_sql_start_transaction();
res=psync_sql_prep_statement("UPDATE pagecache SET type="NTO_STR(PAGE_TYPE_FREE)", hash=NULL, pageid=NULL, crc=NULL WHERE id=?");
for (i=0; i<cnt; i++){
psync_sql_bind_uint(res, 1, entries[i].id);
psync_sql_run(res);
free_db_pages++;
if ((i&0x1f)==0x1f && psync_sql_has_waiters()){
psync_sql_free_result(res);
psync_sql_commit_transaction();
debug(D_NOTICE, "got waiters for sql lock, pausing for a while");
psync_milisleep(5);
psync_sql_start_transaction();
res=psync_sql_prep_statement("UPDATE pagecache SET type="NTO_STR(PAGE_TYPE_FREE)", hash=NULL, pageid=NULL, crc=NULL WHERE id=?");
}
else if ((i&0xfff)==0xfff){
psync_sql_free_result(res);
psync_sql_commit_transaction();
psync_sql_start_transaction();
res=psync_sql_prep_statement("UPDATE pagecache SET type="NTO_STR(PAGE_TYPE_FREE)", hash=NULL, pageid=NULL, crc=NULL WHERE id=?");
}
}
psync_sql_free_result(res);
psync_sql_commit_transaction();
/* ocnt=(cnt+63)/64;
for (j=0; j<ocnt; j++){
i=j*64;
e=i+64;
if (e>cnt)
e=cnt;
psync_sql_start_transaction();
res=psync_sql_prep_statement("UPDATE pagecache SET type="NTO_STR(PAGE_TYPE_FREE)", hash=NULL, pageid=NULL, crc=NULL WHERE id=?");
for (; i<e; i++){
psync_sql_bind_uint(res, 1, entries[i].id);
psync_sql_run(res);
free_db_pages++;
}
psync_sql_free_result(res);
psync_sql_commit_transaction();
psync_milisleep(5);
// 64 pages per 5 millisec is around 50Mb/sec, more than enough
}*/
clean_cache_in_progress=0;
pthread_mutex_unlock(&clean_cache_mutex);
psync_free(oentries);
debug(D_NOTICE, "syncing database");
psync_sql_sync();
debug(D_NOTICE, "finished cleaning cache, free cache pages %u", (unsigned)free_db_pages);
}
static int cmp_flush_pages(const psync_list *p1, const psync_list *p2){
const psync_cache_page_t *page1, *page2;
page1=psync_list_element(p1, const psync_cache_page_t, flushlist);
page2=psync_list_element(p2, const psync_cache_page_t, flushlist);
if (page1->hash<page2->hash)
return -1;
else if (page1->hash>page2->hash)
return 1;
else if (page1->pageid<page2->pageid)
return -1;
else if (page1->pageid>page2->pageid)
return 1;
else
return 0;
}
static int cmp_discard_pages(const psync_list *p1, const psync_list *p2){
const psync_cache_page_t *page1, *page2;
page1=psync_list_element(p1, const psync_cache_page_t, flushlist);
page2=psync_list_element(p2, const psync_cache_page_t, flushlist);
if (page1->lastuse<page2->lastuse)
return -1;
else if (page1->lastuse>page2->lastuse)
return 1;
else
return 0;
}
static int check_disk_full(){
int64_t filesize, freespace;
uint64_t minlocal, maxpage, addspc;
psync_sql_res *res;
db_cache_max_page=psync_sql_cellint("SELECT MAX(id) FROM pagecache", 0);
filesize=psync_file_size(readcache);
if (unlikely_log(filesize==-1))
return 0;
freespace=psync_get_free_space_by_path(psync_setting_get_string(_PS(fscachepath)));
minlocal=psync_setting_get_uint(_PS(minlocalfreespace));
if (unlikely_log(freespace==-1))
return 0;
if (db_cache_max_page*PSYNC_FS_PAGE_SIZE>filesize)
addspc=cache_pages_in_hash*PSYNC_FS_PAGE_SIZE;
else
addspc=0;
if (minlocal+addspc<=freespace){
psync_set_local_full(0);
return 0;
}
debug(D_NOTICE, "local disk is full, freespace=%lu, minfreespace=%lu", (unsigned long)freespace, (unsigned long)minlocal);
psync_set_local_full(1);
if (minlocal>=freespace)
maxpage=filesize/PSYNC_FS_PAGE_SIZE;
else
maxpage=(filesize+freespace-minlocal)/PSYNC_FS_PAGE_SIZE;
res=psync_sql_prep_statement("DELETE FROM pagecache WHERE id>?");
psync_sql_bind_uint(res, 1, maxpage);
psync_sql_run_free(res);
free_db_pages=psync_sql_cellint("SELECT COUNT(*) FROM pagecache WHERE type="NTO_STR(PAGE_TYPE_FREE), 0);
db_cache_max_page=maxpage;
debug(D_NOTICE, "free_db_pages=%u, db_cache_max_page=%lu", (unsigned)free_db_pages, (unsigned long)db_cache_max_page);
return 1;
}
static int flush_pages(int nosleep){
static time_t lastflush=0;
psync_list *l1, *l2;
psync_sql_res *res;
psync_uint_row row;
psync_cache_page_t *page;
psync_list pages_to_flush;
psync_uint_t i, updates, pagecnt;
time_t ctime;
uint32_t cpih;
int ret, diskfull;
pthread_mutex_lock(&cache_mutex);
flush_page_running++;
flushcacherun=1;
pthread_mutex_unlock(&cache_mutex);
flushedbetweentimers=1;
pthread_mutex_lock(&flush_cache_mutex);
diskfull=check_disk_full();
updates=0;
pagecnt=0;
ctime=psync_timer_time();
psync_list_init(&pages_to_flush);
pthread_mutex_lock(&cache_mutex);
if (unlikely(diskfull && free_db_pages==0)){
debug(D_NOTICE, "disk is full, discarding some pages");
for (i=0; i<CACHE_HASH; i++)
psync_list_for_each_safe(l1, l2, &cache_hash[i]){
page=psync_list_element(l1, psync_cache_page_t, list);
if (page->type==PAGE_TYPE_READ)
psync_list_add_tail(&pages_to_flush, &page->flushlist);
else if (page->type==PAGE_TYPE_CACHE){
psync_list_del(&page->list);
psync_list_add_head(&free_pages, &page->list);
cache_pages_in_hash--;
cache_pages_free++;
}
}
pthread_mutex_unlock(&cache_mutex);
psync_list_sort(&pages_to_flush, cmp_discard_pages);
pthread_mutex_lock(&cache_mutex);
i=0;
psync_list_for_each_element(page, &pages_to_flush, psync_cache_page_t, flushlist){
psync_list_del(&page->list);
psync_list_add_head(&free_pages, &page->list);
cache_pages_in_hash--;
cache_pages_free++;
if (++i>=CACHE_PAGES/2)
break;
}
debug(D_NOTICE, "discarded %u pages", (unsigned)i);
psync_list_init(&pages_to_flush);
if (free_page_waiters)
pthread_cond_broadcast(&free_page_cond);
}
if (cache_pages_in_hash){
debug(D_NOTICE, "flushing cache free_db_pages=%u", (unsigned)free_db_pages);
cache_pages_reset=0;
for (i=0; i<CACHE_HASH; i++)
psync_list_for_each_safe(l1, l2, &cache_hash[i]){
page=psync_list_element(l1, psync_cache_page_t, list);
if (page->type==PAGE_TYPE_READ){
psync_list_add_tail(&pages_to_flush, &page->flushlist);
pagecnt++;
}
else if (page->type==PAGE_TYPE_CACHE){
psync_list_del(&page->list);
psync_list_add_head(&free_pages, &page->list);
cache_pages_free++;
}
}
cache_pages_in_hash=pagecnt;
if (pagecnt){
pthread_mutex_unlock(&cache_mutex);
debug(D_NOTICE, "cache_pages_in_hash=%u", (unsigned)pagecnt);
psync_list_sort(&pages_to_flush, cmp_flush_pages);
res=psync_sql_query_rdlock("SELECT id FROM pagecache WHERE type="NTO_STR(PAGE_TYPE_FREE)" ORDER BY id LIMIT ?");
psync_sql_bind_uint(res, 1, pagecnt);
psync_list_for_each_element(page, &pages_to_flush, psync_cache_page_t, flushlist){
if (unlikely(!(row=psync_sql_fetch_rowint(res)))){
psync_list *l1, *l2;
l1=&page->flushlist;
do{
l2=l1->next;
psync_list_del(l1);
l1=l2;
} while (l1!=&pages_to_flush);
break;
}
page->flushpageid=row[0];
}
psync_sql_free_result(res);
/* res=psync_sql_query("SELECT id FROM pagecache WHERE type="NTO_STR(PAGE_TYPE_FREE)" ORDER BY id");
for (i=0; i<CACHE_HASH; i++)
psync_list_for_each_element(page, &cache_hash[i], psync_cache_page_t, list)
if (page->type==PAGE_TYPE_READ){
if (!(row=psync_sql_fetch_rowint(res)))
goto break2;
page->flushpageid=row[0];
psync_list_add_tail(&pages_to_flush, &page->flushlist);
}
break2:
psync_sql_free_result(res);
pthread_mutex_unlock(&cache_mutex);*/
i=0;
psync_list_for_each_element(page, &pages_to_flush, psync_cache_page_t, flushlist){
if (psync_file_pwrite(readcache, page->page, PSYNC_FS_PAGE_SIZE, (uint64_t)page->flushpageid*PSYNC_FS_PAGE_SIZE)!=PSYNC_FS_PAGE_SIZE){
debug(D_ERROR, "write to cache file failed");
pthread_mutex_unlock(&flush_cache_mutex);
return -1;
}
i++;
}
debug(D_NOTICE, "cache data of %u pages written", (unsigned)i);
psync_file_schedulesync(readcache);
/* if we can afford it, wait a while before calling fsync() as at least on Linux this blocks reads from the same file until it returns */
if (nosleep!=1){
if (nosleep==2)
i=180;
else
i=0;
pthread_mutex_lock(&cache_mutex);
while (cache_pages_free>=CACHE_PAGES*5/100 && i++<200){
pthread_mutex_unlock(&cache_mutex);
psync_milisleep(10);
pthread_mutex_lock(&cache_mutex);
}
pthread_mutex_unlock(&cache_mutex);
}
debug(D_NOTICE, "syncing cache data");
if (psync_file_sync(readcache)){
debug(D_ERROR, "flush of cache file failed");
pthread_mutex_unlock(&flush_cache_mutex);
return -1;
}
debug(D_NOTICE, "cache data synced");
pthread_mutex_lock(&cache_mutex);
}
}
pthread_mutex_unlock(&cache_mutex);
psync_sql_start_transaction();
pthread_mutex_lock(&cache_mutex);
if (db_cache_max_page<db_cache_in_pages && cache_pages_in_hash && !diskfull){
i=0;
res=psync_sql_prep_statement("INSERT INTO pagecache (type) VALUES ("NTO_STR(PAGE_TYPE_FREE)")");
while (db_cache_max_page+i<db_cache_in_pages && i<CACHE_PAGES && i<cache_pages_in_hash){
psync_sql_run(res);
i++;
if (i%64==0){
pthread_mutex_unlock(&cache_mutex);
psync_sql_free_result(res);
psync_sql_commit_transaction();
if (!nosleep && !free_page_waiters)
psync_milisleep(1);
psync_sql_start_transaction();
pthread_mutex_lock(&cache_mutex);
res=psync_sql_prep_statement("INSERT INTO pagecache (type) VALUES ("NTO_STR(PAGE_TYPE_FREE)")");
}
}
psync_sql_free_result(res);
free_db_pages+=i;
db_cache_max_page+=i;
debug(D_NOTICE, "inserted %lu new free pages to database, db_cache_in_pages=%lu, db_cache_max_page=%lu",
(unsigned long)i, (unsigned long)db_cache_in_pages, (unsigned long)db_cache_max_page);
updates++;
}
cpih=cache_pages_in_hash;
if (!psync_list_isempty(&pages_to_flush)){
pagecnt=0;
res=psync_sql_prep_statement("UPDATE OR IGNORE pagecache SET hash=?, pageid=?, type="NTO_STR(PAGE_TYPE_READ)", lastuse=?, usecnt=?, size=?, crc=? WHERE id=?");
psync_list_for_each_element(page, &pages_to_flush, psync_cache_page_t, flushlist){
psync_list_del(&page->list);
psync_sql_bind_uint(res, 1, page->hash);
psync_sql_bind_uint(res, 2, page->pageid);
psync_sql_bind_uint(res, 3, page->lastuse);
psync_sql_bind_uint(res, 4, page->usecnt);
psync_sql_bind_uint(res, 5, page->size);
psync_sql_bind_uint(res, 6, page->crc);
psync_sql_bind_uint(res, 7, page->flushpageid);
psync_sql_run(res);
if (likely(psync_sql_affected_rows())){
updates++;
pagecnt++;
free_db_pages--;
}
psync_list_add_head(&free_pages, &page->list);
cache_pages_free++;
if (nosleep!=1 && updates%64==0){
if (free_page_waiters)
pthread_cond_broadcast(&free_page_cond);
pthread_mutex_unlock(&cache_mutex);
psync_sql_free_result(res);
psync_sql_commit_transaction();
if (!free_page_waiters) // it is ok if we read a stale value, because we don't hold cache_mutex any more
psync_milisleep(1);
psync_sql_start_transaction();
pthread_mutex_lock(&cache_mutex);
res=psync_sql_prep_statement("UPDATE OR IGNORE pagecache SET hash=?, pageid=?, type="NTO_STR(PAGE_TYPE_READ)", lastuse=?, usecnt=?, size=?, crc=? WHERE id=?");
}
}
psync_sql_free_result(res);
debug(D_NOTICE, "flushed %u pages to cache file, free db pages %u, cache_pages_in_hash=%u", (unsigned)pagecnt,
(unsigned)free_db_pages, (unsigned)cache_pages_in_hash);
cache_pages_in_hash-=pagecnt;
}
if (cachepages_to_update_cnt && (cpih || cachepages_to_update_cnt>=DB_CACHE_UPDATE_HASH/4 || lastflush+300<ctime)){
res=psync_sql_prep_statement("UPDATE pagecache SET lastuse=?, usecnt=usecnt+? WHERE id=?");
for (i=0; i<DB_CACHE_UPDATE_HASH; i++)
if (cachepages_to_update[i].pagecacheid){
psync_sql_bind_uint(res, 1, cachepages_to_update[i].lastuse);
psync_sql_bind_uint(res, 2, cachepages_to_update[i].usecnt);
psync_sql_bind_uint(res, 3, cachepages_to_update[i].pagecacheid);
psync_sql_run(res);
memset(&cachepages_to_update[i], 0, sizeof(psync_cachepage_to_update));
updates++;
if (!nosleep && updates%128==0){
pthread_mutex_unlock(&cache_mutex);
psync_sql_free_result(res);
psync_sql_commit_transaction();
if (!free_page_waiters)
psync_milisleep(1);
psync_sql_start_transaction();
pthread_mutex_lock(&cache_mutex);
res=psync_sql_prep_statement("UPDATE pagecache SET lastuse=?, usecnt=usecnt+? WHERE id=?");
}
}
psync_sql_free_result(res);
debug(D_NOTICE, "flushed %u access records to database", (unsigned)cachepages_to_update_cnt);
cachepages_to_update_cnt=0;
if (!nosleep)
for (i=0; i<DB_CACHE_UPDATE_HASH; i++)
if (cachepages_to_update[i].pagecacheid)
cachepages_to_update_cnt++;
lastflush=ctime;
}
flushcacherun=0;
flush_page_running--;
if (free_page_waiters){
debug(D_NOTICE, "finished flushing cache, but there are still free page waiters, broadcasting");
pthread_cond_broadcast(&free_page_cond);
}
if (updates){
pthread_mutex_unlock(&cache_mutex);
ret=psync_sql_commit_transaction();
pthread_mutex_unlock(&flush_cache_mutex);
if (free_db_pages<=CACHE_PAGES*2)
psync_run_thread("clean cache", clean_cache);
return ret;
}
else{
pthread_mutex_unlock(&cache_mutex);
psync_sql_rollback_transaction();
pthread_mutex_unlock(&flush_cache_mutex);
if (free_db_pages==0)
psync_run_thread("clean cache", clean_cache);
return 0;
}
}
int psync_pagecache_flush(){
if (flush_pages(1))
return -EIO;
else
return 0;
}
static void psync_pagecache_flush_timer(psync_timer_t timer, void *ptr){
if (!flushedbetweentimers && (cache_pages_in_hash || cachepages_to_update_cnt))
psync_run_thread("flush pages timer", flush_pages_noret);
flushedbetweentimers=0;
pthread_mutex_lock(&cache_mutex);
if (cache_pages_free==CACHE_PAGES && !cache_pages_reset){
cache_pages_reset=1;
debug(D_NOTICE, "resetting free pages");
psync_anon_reset(pages_base, CACHE_PAGES*PSYNC_FS_PAGE_SIZE);
}
pthread_mutex_unlock(&cache_mutex);
}
static void mark_pagecache_used(uint64_t pagecacheid){
uint64_t h;
time_t tm;
int tries, runthread;
h=pagecacheid%DB_CACHE_UPDATE_HASH;
tm=psync_timer_time();
tries=0;
runthread=0;
pthread_mutex_lock(&cache_mutex);
while (++tries<=10){
if (cachepages_to_update[h].pagecacheid==0){
cachepages_to_update[h].pagecacheid=pagecacheid;
cachepages_to_update[h].lastuse=tm;
cachepages_to_update[h].usecnt=1;
cachepages_to_update_cnt++;
break;
}
else if (cachepages_to_update[h].pagecacheid==pagecacheid){
if (tm>cachepages_to_update[h].lastuse+5){
cachepages_to_update[h].lastuse=tm;
cachepages_to_update[h].usecnt++;
}
break;
}
if (++h>=DB_CACHE_UPDATE_HASH)
h=0;
}
if (cachepages_to_update_cnt>DB_CACHE_UPDATE_HASH/2 && !flushcacherun){
flushcacherun=1;
runthread=1;
}
pthread_mutex_unlock(&cache_mutex);
if (runthread)
psync_run_thread("flush pages mark_pagecache_used", flush_pages_noret);
}
static void mark_pagescache_used(uint64_t first_page_id, psync_uint_t pagecnt, unsigned char *dbread){
time_t tm;
psync_uint_t i;
uint64_t h;
int tries, runthread;
runthread=0;
tm=psync_timer_time();
pthread_mutex_lock(&cache_mutex);
for (i=0; i<pagecnt; i++)
if (dbread[i/8]&(1<<(i%8))){
h=(first_page_id+i)%DB_CACHE_UPDATE_HASH;
tries=0;
while (++tries<=10){
if (cachepages_to_update[h].pagecacheid==0){
cachepages_to_update[h].pagecacheid=first_page_id+i;
cachepages_to_update[h].lastuse=tm;
cachepages_to_update[h].usecnt=1;
cachepages_to_update_cnt++;
break;
}
else if (cachepages_to_update[h].pagecacheid==first_page_id+i){
if (tm>cachepages_to_update[h].lastuse+5){
cachepages_to_update[h].lastuse=tm;
cachepages_to_update[h].usecnt++;
}
break;
}
if (++h>=DB_CACHE_UPDATE_HASH)
h=0;
}
}
if (cachepages_to_update_cnt>DB_CACHE_UPDATE_HASH/2 && !flushcacherun){
flushcacherun=1;
runthread=1;
}
pthread_mutex_unlock(&cache_mutex);
if (runthread)
psync_run_thread("flush pages mark_pagescache_used", flush_pages_noret);
}
PSYNC_NOINLINE static void mark_page_free(uint64_t pagecacheid){
psync_sql_res *res;
res=psync_sql_prep_statement("UPDATE pagecache SET type="NTO_STR(PAGE_TYPE_FREE)", pageid=NULL, hash=NULL WHERE id=?");
psync_sql_bind_uint(res, 1, pagecacheid);
psync_sql_run_free(res);
}
static psync_int_t check_page_in_database_by_hash(uint64_t hash, uint64_t pageid, char *buff, psync_uint_t size, psync_uint_t off){
psync_sql_res *res;
psync_uint_row row;
size_t dsize;
ssize_t readret;
psync_int_t ret;
uint64_t pagecacheid;
uint32_t crc;
ret=-1;
res=psync_sql_query_rdlock("SELECT id, size, crc FROM pagecache WHERE type=+"NTO_STR(PAGE_TYPE_READ)" AND hash=? AND pageid=?");
psync_sql_bind_uint(res, 1, hash);
psync_sql_bind_uint(res, 2, pageid);
if ((row=psync_sql_fetch_rowint(res))){
pagecacheid=row[0];
dsize=row[1];
crc=row[2];
if (size+off>dsize){
if (off>dsize)
size=0;
else
size=dsize-off;
}
ret=size;
}
psync_sql_free_result(res);
if (ret!=-1){
readret=psync_file_pread(readcache, buff, size, pagecacheid*PSYNC_FS_PAGE_SIZE+off);
if (unlikely(readret!=size)){
debug(D_ERROR, "failed to read %lu bytes from cache file at offset %lu, read returned %ld, errno=%ld",
(unsigned long)size, (unsigned long)(pagecacheid*PSYNC_FS_PAGE_SIZE+off), (long)readret, (long)psync_fs_err());
mark_page_free(pagecacheid);
ret=-1;
}
else{
if (unlikely(size==dsize && off==0 && psync_crc32c(PSYNC_CRC_INITIAL, buff, size)!=crc)){
debug(D_WARNING, "got bad CRC when reading data from cache at offset %lu", (unsigned long)(pagecacheid*PSYNC_FS_PAGE_SIZE+off));
mark_page_free(pagecacheid);
ret=-1;
}
else
mark_pagecache_used(pagecacheid);
}
}
return ret;
}
static void check_pages_in_database_by_hash(uint64_t hash, uint64_t first_page_id, psync_uint_t pagecnt, char *buff, unsigned char *dbread){
psync_sql_res *res;
psync_full_result_int *fres;
uint64_t cid, cpid;
ssize_t readret;
uint32_t i, j, cnt;
res=psync_sql_query_rdlock("SELECT id, pageid, size, crc FROM pagecache WHERE type=+"NTO_STR(PAGE_TYPE_READ)
" AND hash=? AND pageid>=? AND pageid<? ORDER BY pageid");
psync_sql_bind_uint(res, 1, hash);
psync_sql_bind_uint(res, 2, first_page_id);
psync_sql_bind_uint(res, 3, first_page_id+pagecnt);
fres=psync_sql_fetchall_int(res);
cnt=1;
for (i=0; i<fres->rows; i+=cnt){
if (psync_get_result_cell(fres, i, 2)!=PSYNC_FS_PAGE_SIZE)
continue;
cid=psync_get_result_cell(fres, i, 0);
cpid=psync_get_result_cell(fres, i, 1);
cnt=0;
while (i+cnt+1<fres->rows && psync_get_result_cell(fres, i+cnt+1, 0)==cid+cnt+1 && psync_get_result_cell(fres, i+cnt+1, 1)==cpid+cnt+1 &&
psync_get_result_cell(fres, i+cnt+1, 2)==PSYNC_FS_PAGE_SIZE)
cnt++;
cnt++;
// debug(D_NOTICE, "reading %u consecutive pages from cache file id %lu, firstpageid %lu", (unsigned)cnt, (unsigned long)cid, (unsigned long)cpid);
readret=psync_file_pread(readcache, buff+(cpid-first_page_id)*PSYNC_FS_PAGE_SIZE, PSYNC_FS_PAGE_SIZE*cnt, cid*PSYNC_FS_PAGE_SIZE);
if (readret!=PSYNC_FS_PAGE_SIZE*cnt){
debug(D_ERROR, "failed to read %lu bytes from cache file at offset %lu, read returned %ld, errno=%ld",
(unsigned long)(PSYNC_FS_PAGE_SIZE*cnt), (unsigned long)(cid*PSYNC_FS_PAGE_SIZE), (long)readret, (long)psync_fs_err());
continue;
}
for (j=0; j<cnt; j++)
if (psync_crc32c(PSYNC_CRC_INITIAL, buff+(cpid-first_page_id+j)*PSYNC_FS_PAGE_SIZE, PSYNC_FS_PAGE_SIZE)==psync_get_result_cell(fres, i+j, 3))
dbread[(cpid-first_page_id+j)/8]|=1<<((cpid-first_page_id+j)%8);
else
debug(D_WARNING, "got bad CRC when reading data from cache at offset %lu", (unsigned long)((cid+j)*PSYNC_FS_PAGE_SIZE));
}
if (fres->rows)
mark_pagescache_used(first_page_id, pagecnt, dbread);
psync_free(fres);
}
static psync_int_t check_page_in_database_by_hash_and_cache(uint64_t hash, uint64_t pageid, char *buff, psync_uint_t size, psync_uint_t off){
psync_sql_res *res;
psync_uint_row row;
psync_cache_page_t *page;
size_t dsize;
ssize_t readret;
psync_int_t ret;
uint64_t pagecacheid;
uint32_t crc, ccrc;
ret=-1;
res=psync_sql_query_rdlock("SELECT id, size, crc FROM pagecache WHERE type=+"NTO_STR(PAGE_TYPE_READ)" AND hash=? AND pageid=?");
psync_sql_bind_uint(res, 1, hash);
psync_sql_bind_uint(res, 2, pageid);
if ((row=psync_sql_fetch_rowint(res))){
pagecacheid=row[0];
dsize=row[1];
crc=row[2];
if (size+off>dsize){
if (off>dsize)
size=0;
else
size=dsize-off;
}
ret=size;
}
psync_sql_free_result(res);
if (ret!=-1){
page=psync_pagecache_get_free_page(0);
readret=psync_file_pread(readcache, page->page, dsize, pagecacheid*PSYNC_FS_PAGE_SIZE);
if (unlikely(readret!=dsize)){
debug(D_ERROR, "failed to read %lu bytes from cache file at offset %lu, read returned %ld, errno=%ld",
(unsigned long)dsize, (unsigned long)(pagecacheid*PSYNC_FS_PAGE_SIZE), (long)readret, (long)psync_fs_err());
mark_page_free(pagecacheid);
psync_pagecache_return_free_page(page);
ret=-1;
}
else{
ccrc=psync_crc32c(PSYNC_CRC_INITIAL, page->page, dsize);
if (unlikely(ccrc!=crc)){
debug(D_WARNING, "got bad CRC when reading data from cache at offset %lu, size %lu db CRC %u calculated CRC %u",
(unsigned long)(pagecacheid*PSYNC_FS_PAGE_SIZE), (unsigned long)dsize, (unsigned)crc, (unsigned)ccrc);
mark_page_free(pagecacheid);
psync_pagecache_return_free_page(page);
ret=-1;
}
else{
mark_pagecache_used(pagecacheid);
memcpy(buff, page->page+off, size);
page->hash=hash;
page->pageid=pageid;
page->lastuse=0;
page->size=dsize;
page->usecnt=0;
page->crc=ccrc;
page->type=PAGE_TYPE_CACHE;
pthread_mutex_lock(&cache_mutex);
psync_list_add_tail(&cache_hash[pagehash_by_hash_and_pageid(hash, pageid)], &page->list);
cache_pages_in_hash++;
pthread_mutex_unlock(&cache_mutex);
}
}
}
return ret;
}
int psync_pagecache_read_modified_locked(psync_openfile_t *of, char *buf, uint64_t size, uint64_t offset){
psync_interval_tree_t *fi;
uint64_t isize, ioffset;
ssize_t br;
int rd;
fi=psync_interval_tree_first_interval_containing_or_after(of->writeintervals, offset);
if (fi && fi->from<=offset && fi->to>=offset+size){
debug(D_NOTICE, "reading %lu bytes at offset %lu only from local storage", (unsigned long)size, (unsigned long)offset);
br=psync_file_pread(of->datafile, buf, size, offset);
pthread_mutex_unlock(&of->mutex);
if (br==-1)
return -EIO;
else
return br;
}
rd=psync_pagecache_read_unmodified_locked(of, buf, size, offset);
if (rd<0)
return rd;
psync_fs_lock_file(of);
fi=psync_interval_tree_first_interval_containing_or_after(of->writeintervals, offset);
if (!fi || fi->from>=offset+size){
pthread_mutex_unlock(&of->mutex);
if (fi)
br=fi->from;
else
br=-1;
debug(D_NOTICE, "reading %lu bytes at offset %lu only from remote fileid %lu revision %lu, read returned %d, next local interval starts at %ld",
(unsigned long)size, (unsigned long)offset, (unsigned long)of->remotefileid, (unsigned long)of->hash, rd, (long)br);
return rd;
}
debug(D_NOTICE, "reading %lu bytes at offset %lu from both network and local", (unsigned long)size, (unsigned long)offset);
do {
ioffset=fi->from;
isize=fi->to-fi->from;
if (ioffset<offset){
isize-=offset-ioffset;
ioffset=offset;
}
if (ioffset+isize>offset+size)
isize=offset+size-ioffset;
debug(D_NOTICE, "reading %lu bytes at offset %lu from local storage", (unsigned long)isize, (unsigned long)ioffset);
br=psync_file_pread(of->datafile, buf+ioffset-offset, isize, ioffset);
if (br==-1){
pthread_mutex_unlock(&of->mutex);
return -EIO;
}
if (rd!=size && br+ioffset-offset>rd)
rd=br+ioffset-offset;
fi=psync_interval_tree_get_next(fi);
} while (fi && fi->from<offset+size);
pthread_mutex_unlock(&of->mutex);
return rd;
}
static void psync_pagecache_free_request(psync_request_t *request){
psync_list_for_each_element_call(&request->ranges, psync_request_range_t, list, psync_free);
psync_free(request);
}
static void psync_pagecache_send_error_page_wait(psync_page_wait_t *pw, int err){
psync_page_waiter_t *pwt;
psync_list_del(&pw->list);
psync_list_for_each_element(pwt, &pw->waiters, psync_page_waiter_t, listpage){
pwt->error=err;
pwt->ready=1;
pthread_cond_broadcast(&pwt->cond);
}
psync_free(pw);
}
static void psync_pagecache_send_range_error(psync_request_range_t *range, psync_request_t *request, int err){
uint64_t first_page_id;
psync_page_wait_t *pw;
psync_uint_t len, i, h;
first_page_id=range->offset/PSYNC_FS_PAGE_SIZE;
len=range->length/PSYNC_FS_PAGE_SIZE;
debug(D_NOTICE, "sending error %d to request for offset %lu, length %lu of fileid %lu hash %lu",
err, (unsigned long)range->offset, (unsigned long)range->length, (unsigned long)request->fileid, (unsigned long)request->hash);
for (i=0; i<len; i++){
h=waiterhash_by_hash_and_pageid(request->hash, first_page_id+i);
psync_list_for_each_element(pw, &wait_page_hash[h], psync_page_wait_t, list)
if (pw->hash==request->hash && pw->pageid==first_page_id+i){
psync_pagecache_send_error_page_wait(pw, err);
break;
}
}
}
static void psync_pagecache_send_error(psync_request_t *request, int err){
psync_request_range_t *range;
lock_wait(request->hash);
psync_list_for_each_element(range, &request->ranges, psync_request_range_t, list)
psync_pagecache_send_range_error(range, request, err);
unlock_wait(request->ash);
if (request->needkey)
psync_pagecache_set_bad_encoder(request->of);
psync_fs_dec_of_refcnt_and_readers(request->of);
psync_pagecache_free_request(request);
}
#if IS_DEBUG
#define psync_pagecache_send_error(r, e) do {psync_pagecache_send_error(r, e); debug(D_NOTICE, "sending request error %d", e);} while (0)
#endif
static int psync_pagecache_read_range_from_sock(psync_request_t *request, psync_request_range_t *range, psync_http_socket *sock){
uint64_t first_page_id;
psync_page_wait_t *pw;
psync_cache_page_t *page;
psync_uint_t len, i, h;
int rb;
first_page_id=range->offset/PSYNC_FS_PAGE_SIZE;
len=range->length/PSYNC_FS_PAGE_SIZE;
rb=psync_http_next_request(sock);
if (unlikely(rb)){
if (rb==410 || rb==404 || rb==-1){
debug(D_WARNING, "got %d from psync_http_next_request, freeing URLs and requesting retry, range from %lu", rb, (long unsigned)range->offset);
return 1;
}
else{
debug(D_WARNING, "got %d from psync_http_next_request, returning error", rb);
return -1;
}
}
for (i=0; i<len; i++){
page=psync_pagecache_get_free_page(0);
rb=psync_http_request_readall(sock, page->page, PSYNC_FS_PAGE_SIZE);
if (unlikely_log(rb<=0)){
psync_pagecache_return_free_page(page);
psync_timer_notify_exception();
return -1;
}
page->hash=request->hash;
page->pageid=first_page_id+i;
page->lastuse=psync_timer_time();
page->size=rb;
page->usecnt=0;
page->crc=psync_crc32c(PSYNC_CRC_INITIAL, page->page, rb);
page->type=PAGE_TYPE_READ;
h=waiterhash_by_hash_and_pageid(page->hash, page->pageid);
lock_wait(page->hash);
psync_list_for_each_element(pw, &wait_page_hash[h], psync_page_wait_t, list)
if (pw->hash==page->hash && pw->pageid==page->pageid){
psync_pagecache_send_page_wait_page(pw, page);
break;
}
unlock_wait(page->hash);
pthread_mutex_lock(&cache_mutex);
psync_list_add_tail(&cache_hash[pagehash_by_hash_and_pageid(page->hash, page->pageid)], &page->list);
cache_pages_in_hash++;
pthread_mutex_unlock(&cache_mutex);
}
return 0;
}
//X-Range-Next:
#define PSYNC_CONSTRUCT_ADD_NUMBER(x)\
do {\
if (off<=24)\
return NULL;\
do {\
buff[off--]='0'+x%10;\
x/=10;\
} while (x);\
} while (0)
#define PSYNC_CONSTRUCT_HEADER "X-Range-Next:"
#define PSYNC_DWLTAG_HEADER "Cookie: dwltag="
char *psync_http_construct_range_next_header(psync_request_t *request, psync_urls_t *urls){
char buff[2048];
const binresult *dwltag;
uint64_t nm;
psync_request_range_t *range;
psync_list *l;
size_t off;
l=request->ranges.prev;
off=sizeof(buff)-1;
buff[off--]=0;
buff[off--]='\012';
buff[off--]='\015';
dwltag=psync_find_result(urls->urls, "dwltag", PARAM_STR);
if (dwltag->length<=40 && dwltag->length>0) {
off-=dwltag->length-1;
memcpy(buff+off, dwltag->str, dwltag->length);
off-=sizeof(PSYNC_DWLTAG_HEADER)-1;
memcpy(buff+off, PSYNC_DWLTAG_HEADER, sizeof(PSYNC_DWLTAG_HEADER)-1);
off--;
buff[off--]='\012';
buff[off--]='\015';
}
while (l->prev!=&request->ranges){
range=psync_list_element(l, psync_request_range_t, list);
nm=range->offset+range->length-1;
PSYNC_CONSTRUCT_ADD_NUMBER(nm);
buff[off--]='-';
nm=range->offset;
PSYNC_CONSTRUCT_ADD_NUMBER(nm);
buff[off--]=' ';
buff[off--]=',';
l=l->prev;
}
if (off<sizeof(PSYNC_CONSTRUCT_HEADER))
return NULL;
return psync_strdup((char *)memcpy(buff+off-sizeof(PSYNC_CONSTRUCT_HEADER)+3, PSYNC_CONSTRUCT_HEADER, sizeof(PSYNC_CONSTRUCT_HEADER)-1));
}
static void psync_pagecache_read_unmodified_thread(void *ptr){
psync_request_t *request;
psync_http_socket *sock;
psync_socket *api;
const char *host;
const char *path;
char cookie[128];
psync_request_range_t *range;
const binresult *hosts;
psync_urls_t *urls;
psync_crypto_aes256_sector_encoder_decoder_t enc;
int err, tries;
request=(psync_request_t *)ptr;
if (psync_status_get(PSTATUS_TYPE_ONLINE)==PSTATUS_ONLINE_OFFLINE){
psync_pagecache_send_error(request, -ENOTCONN);
return;
}
range=psync_list_element(request->ranges.next, psync_request_range_t, list);
debug(D_NOTICE, "thread run, first offset %lu, size %lu", (unsigned long)range->offset, (unsigned long)range->length);
tries=0;
retry:
if (!(urls=get_urls_for_request(request))){
psync_pagecache_send_error(request, -EIO);
return;
}
if (unlikely(request->needkey)){
enc=psync_cloud_crypto_get_file_encoder(request->fileid, request->hash, 0);
if (psync_crypto_to_error(enc)){
psync_pagecache_send_error(request, -EIO);
return;
}
psync_fs_lock_file(request->of);
if (likely_log(request->of->encoder==PSYNC_CRYPTO_LOADING_SECTOR_ENCODER)){
request->of->encoder=enc;
pthread_cond_broadcast(&enc_key_cond);
}
else
psync_cloud_crypto_release_file_encoder(request->fileid, request->hash, enc);
pthread_mutex_unlock(&request->of->mutex);
request->needkey=0;
}
if (psync_list_isempty(&request->ranges)){
release_urls(urls);
psync_fs_dec_of_refcnt_and_readers(request->of);
psync_pagecache_free_request(request);
return;
}
hosts=psync_find_result(urls->urls, "hosts", PARAM_ARRAY);
psync_slprintf(cookie, sizeof(cookie), "Cookie: dwltag=%s\015\012", psync_find_result(urls->urls, "dwltag", PARAM_STR)->str);
sock=psync_http_connect_multihost_from_cache(hosts, &host);
if (!sock){
if ((api=psync_apipool_get_from_cache())){
debug(D_NOTICE, "no cached server connections, but got cached API connection, serving request from API");
if (likely_log(hosts->length && hosts->array[0]->type==PARAM_STR))
psync_http_connect_and_cache_host(hosts->array[0]->str);
psync_socket_set_write_buffered(api);
psync_list_for_each_element(range, &request->ranges, psync_request_range_t, list){
debug(D_NOTICE, "sending request for offset %lu, size %lu to API", (unsigned long)range->offset, (unsigned long)range->length);
if (psync_api_send_read_request(api, request->fileid, request->hash, range->offset, range->length))
goto err_api1;
}
mark_api_shared(api);
psync_list_for_each_element(range, &request->ranges, psync_request_range_t, list)
if ((err=psync_pagecache_read_range_from_api(request, range, api))){
mark_shared_api_bad(api);
if (err==-2 && psync_list_is_head(&request->ranges, &range->list))
goto err_api1;
else{
psync_apipool_release_bad(api);
goto err0;
}
}
if (pass_shared_api(api)){
psync_socket_clear_write_buffered(api);
psync_apipool_release(api);
}
debug(D_NOTICE, "request from API finished");
goto ok1;
err_api1:
psync_socket_clear_write_buffered_thread(api);
psync_apipool_release_bad(api);
debug(D_WARNING, "error reading range from API, trying from content servers");
}
else if ((api=get_shared_api())){
psync_socket_set_write_buffered_thread(api);
debug(D_NOTICE, "no cached server connections, no cached API servers, but got shared API connection sending request to shared API");
psync_list_for_each_element(range, &request->ranges, psync_request_range_t, list){
debug(D_NOTICE, "sending request for offset %lu, size %lu to shared API", (unsigned long)range->offset, (unsigned long)range->length);
if (psync_api_send_read_request_thread(api, request->fileid, request->hash, range->offset, range->length))
goto err_api2;
}
psync_socket_try_write_buffer_thread(api);
if (wait_shared_api())
goto err_api0;
psync_list_for_each_element(range, &request->ranges, psync_request_range_t, list)
if ((err=psync_pagecache_read_range_from_api(request, range, api))){
mark_shared_api_bad(api);
if (err==-2 && psync_list_is_head(&request->ranges, &range->list))
goto err_api0;
else{
psync_apipool_release_bad(api);
goto err0;
}
}
if (pass_shared_api(api)){
psync_socket_clear_write_buffered(api);
psync_apipool_release(api);
}
debug(D_NOTICE, "request from shared API finished");
goto ok1;
err_api2:
psync_socket_clear_write_buffered_thread(api);
release_bad_shared_api(api);
err_api0:
debug(D_WARNING, "error reading range from API, trying from content servers");
}
}
if (!sock)
sock=psync_http_connect_multihost(hosts, &host);
if (unlikely_log(!sock))
goto err0;
// debug(D_NOTICE, "connected to %s", host);
path=psync_find_result(urls->urls, "path", PARAM_STR)->str;
psync_socket_set_write_buffered(sock->sock);
psync_list_for_each_element(range, &request->ranges, psync_request_range_t, list){
debug(D_NOTICE, "sending request for offset %lu, size %lu", (unsigned long)range->offset, (unsigned long)range->length);
if (psync_list_is_head(&request->ranges, &range->list) && !psync_list_is_tail(&request->ranges, &range->list)){
char *range_hdr=psync_http_construct_range_next_header(request, urls);
debug(D_NOTICE, "sending additional header: %s", range_hdr);
err=psync_http_request_range_additional(sock, host, path, range->offset, range->offset+range->length-1, range_hdr);
psync_free(range_hdr);
}
else
err=psync_http_request(sock, host, path, range->offset, range->offset+range->length-1, cookie);
if (err){
if (tries++<5){
psync_http_close(sock);
goto retry;
}
else
goto err1;
}
}
psync_list_for_each_element(range, &request->ranges, psync_request_range_t, list)
if ((err=psync_pagecache_read_range_from_sock(request, range, sock))){
if (err==1 && tries++<5){
psync_http_close(sock);
release_bad_urls(urls);
goto retry;
}
else
goto err1;
}
psync_socket_clear_write_buffered(sock->sock);
psync_http_close(sock);
debug(D_NOTICE, "request from %s finished", host);
ok1:
psync_fs_dec_of_refcnt_and_readers(request->of);
psync_pagecache_free_request(request);
release_urls(urls);
return;
err1:
psync_http_close(sock);
err0:
psync_pagecache_send_error(request, -EIO);
release_urls(urls);
return;
}
static void check_or_request_page(uint64_t fileid, uint64_t hash, uint64_t pageid, psync_list *ranges){
if (has_page_in_cache_by_hash(hash, pageid) || has_page_in_db(hash, pageid))
return;
else{
psync_page_wait_t *pw;
psync_request_range_t *range;
psync_int_t h;
int found;
h=waiterhash_by_hash_and_pageid(hash, pageid);
found=0;
lock_wait(hash);
psync_list_for_each_element(pw, &wait_page_hash[h], psync_page_wait_t, list)
if (pw->hash==hash && pw->pageid==pageid){
found=1;
break;
}
if (!found){
pw=psync_new(psync_page_wait_t);
psync_list_add_tail(&wait_page_hash[h], &pw->list);
psync_list_init(&pw->waiters);
pw->hash=hash;
pw->pageid=pageid;
pw->fileid=fileid;
if (psync_list_isempty(ranges))
range=NULL;
else
range=psync_list_element(ranges->prev, psync_request_range_t, list);
if (range && range->offset+range->length==pageid*PSYNC_FS_PAGE_SIZE)
range->length+=PSYNC_FS_PAGE_SIZE;
else{
range=psync_new(psync_request_range_t);
psync_list_add_tail(ranges, &range->list);
range->offset=pageid*PSYNC_FS_PAGE_SIZE;
range->length=PSYNC_FS_PAGE_SIZE;
}
}
unlock_wait(hash);
}
}
static void psync_pagecache_read_unmodified_readahead(psync_openfile_t *of, uint64_t offset, uint64_t size, psync_list *ranges,
psync_fileid_t fileid, uint64_t hash, uint64_t initialsize, psync_crypto_offsets_t *offsets){
uint64_t readahead, frompageoff, topageoff, first_page_id, rto;
psync_int_t i, pagecnt, h, streamid;
psync_page_wait_t *pw;
psync_request_range_t *range;
time_t ctime;
unsigned char *pages_in_db;
int found;
if (offset+size>=initialsize)
return;
readahead=0;
frompageoff=offset/PSYNC_FS_PAGE_SIZE;
topageoff=((offset+size+PSYNC_FS_PAGE_SIZE-1)/PSYNC_FS_PAGE_SIZE)-1;
ctime=psync_timer_time();
found=0;
for (streamid=0; streamid<PSYNC_FS_FILESTREAMS_CNT; streamid++)
if (of->streams[streamid].frompage<=frompageoff && of->streams[streamid].topage+2>=frompageoff){
of->streams[streamid].id=++of->laststreamid;
readahead=of->streams[streamid].length;
of->streams[streamid].frompage=frompageoff;
of->streams[streamid].topage=topageoff;
of->streams[streamid].length+=size;
of->streams[streamid].lastuse=ctime;
break;
}
else if (of->streams[streamid].lastuse>=ctime-2)
found++;
if (streamid==PSYNC_FS_FILESTREAMS_CNT){
uint64_t min;
debug(D_NOTICE, "ran out of readahead streams");
min=~(uint64_t)0;
streamid=0;
for (i=0; i<PSYNC_FS_FILESTREAMS_CNT; i++)
if (of->streams[i].id<min){
min=of->streams[i].id;
streamid=i;
}
of->streams[streamid].id=++of->laststreamid;
of->streams[streamid].frompage=frompageoff;
of->streams[streamid].topage=topageoff;
of->streams[streamid].length=size;
of->streams[streamid].requestedto=0;
of->streams[streamid].lastuse=ctime;
if (found==1 && of->currentspeed*4>readahead && !psync_list_isempty(ranges)){
debug(D_NOTICE, "found just one freshly used stream, increasing readahead to four times current speed %u", (unsigned int)of->currentspeed*4);
readahead=size_round_up_to_page(of->currentspeed*4);
}
}
if (of->runningreads>=6 && psync_list_isempty(ranges))
return;
if (offset==0 && (size<PSYNC_FS_MIN_READAHEAD_START) && readahead<PSYNC_FS_MIN_READAHEAD_START-size)
readahead=PSYNC_FS_MIN_READAHEAD_START-size;
else if (offset==PSYNC_FS_MIN_READAHEAD_START/2 && readahead==PSYNC_FS_MIN_READAHEAD_START/2){
of->streams[streamid].length+=offset;
readahead=(PSYNC_FS_MIN_READAHEAD_START/2)*3;
}
else if (offset!=0 && (size<PSYNC_FS_MIN_READAHEAD_RAND) && readahead<PSYNC_FS_MIN_READAHEAD_RAND-size)
readahead=PSYNC_FS_MIN_READAHEAD_RAND-size;
if (of->currentspeed*PSYNC_FS_MAX_READAHEAD_SEC>PSYNC_FS_MIN_READAHEAD_START){
if (readahead>of->currentspeed*PSYNC_FS_MAX_READAHEAD_SEC)
readahead=size_round_up_to_page(of->currentspeed*PSYNC_FS_MAX_READAHEAD_SEC);
if (readahead>PSYNC_FS_MAX_READAHEAD_IF_SEC)
readahead=PSYNC_FS_MAX_READAHEAD_IF_SEC;
}
else if (readahead>PSYNC_FS_MAX_READAHEAD)
readahead=PSYNC_FS_MAX_READAHEAD;
if (psync_list_isempty(ranges)){
if (readahead>=8192*1024)
readahead=(readahead+offset+size)/(4*1024*1024)*(4*1024*1024)-offset-size;
else if (readahead>=2048*1024)
readahead=(readahead+offset+size)/(1024*1024)*(1024*1024)-offset-size;
else if (readahead>=512*1024)
readahead=(readahead+offset+size)/(256*1024)*(256*1024)-offset-size;
else if (readahead>=128*1024)
readahead=(readahead+offset+size)/(64*1024)*(64*1024)-offset-size;
}
if (offset+size+readahead>initialsize)
readahead=size_round_up_to_page(initialsize-offset-size);
rto=of->streams[streamid].requestedto;
if (of->streams[streamid].lastuse<ctime-30)
rto=0;
if (rto<offset+size+readahead)
of->streams[streamid].requestedto=offset+size+readahead;
// debug(D_NOTICE, "rto=%lu", rto);
if (rto>offset+size){
if (rto>offset+size+readahead)
return;
first_page_id=rto/PSYNC_FS_PAGE_SIZE;
pagecnt=(offset+size+readahead-rto)/PSYNC_FS_PAGE_SIZE;
}
else{
first_page_id=(offset+size)/PSYNC_FS_PAGE_SIZE;
pagecnt=readahead/PSYNC_FS_PAGE_SIZE;
}
if (of->encrypted){
uint64_t aoffset, pageid;
psync_int_t l;
uint32_t asize, aoff;
for (i=0; i<pagecnt; i+=PSYNC_CRYPTO_HASH_TREE_SECTORS){
for (l=0; l<offsets->treelevels; l++){
psync_fs_crypto_get_auth_sector_off(first_page_id+i, l, offsets, &aoffset, &asize, &aoff);
pageid=aoffset/PSYNC_FS_PAGE_SIZE;
check_or_request_page(fileid, hash, pageid, ranges);
aoff=aoffset%PSYNC_FS_PAGE_SIZE;
if (aoff && aoff+asize>PSYNC_FS_PAGE_SIZE)
check_or_request_page(fileid, hash, pageid+1, ranges);
}
}
// this may include some auth sectors, but should do no harm
rto=psync_fs_crypto_data_sectorid_by_sectorid(first_page_id+pagecnt);
first_page_id=psync_fs_crypto_data_sectorid_by_sectorid(first_page_id);
pagecnt=rto-first_page_id;
}
pages_in_db=has_pages_in_db(hash, first_page_id, pagecnt, 1);
lock_wait(hash);
for (i=0; i<pagecnt; i++){
if (pages_in_db[i])
continue;
if (has_page_in_cache_by_hash(hash, first_page_id+i))
continue;
h=waiterhash_by_hash_and_pageid(hash, first_page_id+i);
found=0;
psync_list_for_each_element(pw, &wait_page_hash[h], psync_page_wait_t, list)
if (pw->hash==hash && pw->pageid==first_page_id+i){
found=1;
break;
}
if (found)
continue;
// debug(D_NOTICE, "read-aheading page %lu", first_page_id+i);
pw=psync_new(psync_page_wait_t);
psync_list_add_tail(&wait_page_hash[h], &pw->list);
psync_list_init(&pw->waiters);
pw->hash=hash;
pw->pageid=first_page_id+i;
pw->fileid=fileid;
if (psync_list_isempty(ranges))
range=NULL;
else
range=psync_list_element(ranges->prev, psync_request_range_t, list);
if (range && range->offset+range->length==(first_page_id+i)*PSYNC_FS_PAGE_SIZE)
range->length+=PSYNC_FS_PAGE_SIZE;
else{
range=psync_new(psync_request_range_t);
psync_list_add_tail(ranges, &range->list);
range->offset=(first_page_id+i)*PSYNC_FS_PAGE_SIZE;
range->length=PSYNC_FS_PAGE_SIZE;
}
}
unlock_wait(hash);
psync_free(pages_in_db);
if (!psync_list_isempty(ranges))
debug(D_NOTICE, "readahead=%lu, rto=%lu, offset=%lu, size=%lu, currentspeed=%u",
(long unsigned)readahead, (unsigned long)rto, (unsigned long)offset, (unsigned long)size, (unsigned)of->currentspeed);
}
static void psync_free_page_waiter(psync_page_waiter_t *pwt){
pthread_cond_destroy(&pwt->cond);
psync_free(pwt);
}
static psync_page_waiter_t *add_page_waiter(psync_list *wait_list, psync_list *range_list, uint64_t hash, uint64_t pageid, uint64_t fileid,
char *buff, uint32_t pageidx, uint32_t copyoff, uint32_t copysize){
psync_page_waiter_t *pwt;
psync_page_wait_t *pw;
psync_request_range_t *range;
psync_uint_t h;
pwt=psync_new(psync_page_waiter_t);
pthread_cond_init(&pwt->cond, NULL);
pwt->buff=buff;
pwt->pageidx=pageidx;
pwt->off=copyoff;
pwt->size=copysize;
pwt->error=0;
pwt->ready=0;
psync_list_add_tail(wait_list, &pwt->listwaiter);
h=waiterhash_by_hash_and_pageid(hash, pageid);
psync_list_for_each_element(pw, &wait_page_hash[h], psync_page_wait_t, list)
if (pw->hash==hash && pw->pageid==pageid)
goto found;
debug(D_NOTICE, "page %lu not found", (unsigned long)pageid);
pw=psync_new(psync_page_wait_t);
psync_list_add_tail(&wait_page_hash[h], &pw->list);
psync_list_init(&pw->waiters);
pw->hash=hash;
pw->pageid=pageid;
pw->fileid=fileid;
if (psync_list_isempty(range_list))
range=NULL;
else
range=psync_list_element(range_list->prev, psync_request_range_t, list);
if (range && range->offset+range->length==pageid*PSYNC_FS_PAGE_SIZE)
range->length+=PSYNC_FS_PAGE_SIZE;
else{
range=psync_new(psync_request_range_t);
psync_list_add_tail(range_list, &range->list);
range->offset=pageid*PSYNC_FS_PAGE_SIZE;
range->length=PSYNC_FS_PAGE_SIZE;
}
found:
psync_list_add_tail(&pw->waiters, &pwt->listpage);
pwt->waiting_for=pw;
return pwt;
}
static void wait_waiter(psync_page_waiter_t *pwt, uint64_t hash, const char *pt){
lock_wait(hash);
while (!pwt->ready){
debug(D_NOTICE, "waiting for %s page #%lu to be read", pt, (unsigned long)pwt->waiting_for->pageid);
pthread_cond_wait(&pwt->cond, &wait_page_mutex);
debug(D_NOTICE, "waited for %s page", pt); // not safe to use pwt->waiting_for here
}
unlock_wait(hash);
if (pwt->error)
debug(D_WARNING, "reading of page failed with error %d", pwt->error);
}
int psync_pagecache_read_unmodified_locked(psync_openfile_t *of, char *buf, uint64_t size, uint64_t offset){
unsigned char dbread[256];
uint64_t poffset, psize, first_page_id, initialsize, hash;
psync_uint_t pageoff, pagecnt, i, copysize, copyoff;
psync_fileid_t fileid;
psync_int_t rb;
char *pbuff;
psync_page_waiter_t *pwt;
psync_request_t *rq;
psync_list waiting;
int ret;
initialsize=of->initialsize;
hash=of->hash;
fileid=of->remotefileid;
pthread_mutex_unlock(&of->mutex);
if (offset>=initialsize)
return 0;
if (offset+size>initialsize)
size=initialsize-offset;
poffset=offset_round_down_to_page(offset);
pageoff=offset-poffset;
psize=size_round_up_to_page(size+pageoff);
pagecnt=psize/PSYNC_FS_PAGE_SIZE;
first_page_id=poffset/PSYNC_FS_PAGE_SIZE;
psync_list_init(&waiting);
rq=psync_new(psync_request_t);
psync_list_init(&rq->ranges);
lock_wait(hash);
if (pagecnt>1 && pagecnt<=sizeof(dbread)*8 && psize==size){
ret=1;
memset(dbread, 0, sizeof(dbread));
check_pages_in_database_by_hash(hash, first_page_id, pagecnt, buf, dbread);
}
else
ret=0;
for (i=0; i<pagecnt; i++){
if (ret && (dbread[i/8]&(1<<(i%8))))
continue;
if (i==0){
copyoff=pageoff;
if (size>PSYNC_FS_PAGE_SIZE-copyoff)
copysize=PSYNC_FS_PAGE_SIZE-copyoff;
else
copysize=size;
pbuff=buf;
}
else if (i==pagecnt-1){
copyoff=0;
copysize=(size+pageoff)&(PSYNC_FS_PAGE_SIZE-1);
if (!copysize)
copysize=PSYNC_FS_PAGE_SIZE;
pbuff=buf+i*PSYNC_FS_PAGE_SIZE-pageoff;
}
else{
copyoff=0;
copysize=PSYNC_FS_PAGE_SIZE;
pbuff=buf+i*PSYNC_FS_PAGE_SIZE-pageoff;
}
rb=check_page_in_memory_by_hash(hash, first_page_id+i, pbuff, copysize, copyoff);
if (rb==-1 && !ret)
rb=check_page_in_database_by_hash(hash, first_page_id+i, pbuff, copysize, copyoff);
if (rb!=-1){
if (rb==copysize)
continue;
else{
if (i)
size=i*PSYNC_FS_PAGE_SIZE+rb-pageoff;
else
size=rb;
break;
}
}
add_page_waiter(&waiting, &rq->ranges, hash, first_page_id+i, fileid, pbuff, i, copyoff, copysize);
}
unlock_wait(hash);
psync_pagecache_read_unmodified_readahead(of, poffset, psize, &rq->ranges, fileid, hash, initialsize, NULL);
if (!psync_list_isempty(&rq->ranges)){
rq->of=of;
rq->fileid=fileid;
rq->hash=hash;
rq->needkey=0;
psync_fs_inc_of_refcnt_and_readers(of);
psync_run_thread1("read unmodified", psync_pagecache_read_unmodified_thread, rq);
}
else
psync_free(rq);
ret=size;
if (!psync_list_isempty(&waiting)){
psync_list_for_each_element(pwt, &waiting, psync_page_waiter_t, listwaiter){
wait_waiter(pwt, hash, "data");
if (pwt->error)
ret=pwt->error;
else if (pwt->rsize<pwt->size && ret>=0){
if (pwt->rsize){
if (pwt->pageidx)
ret=pwt->pageidx*PSYNC_FS_PAGE_SIZE+pwt->rsize-pageoff;
else
ret=pwt->rsize;
}
else{
if (pwt->pageidx){
if (pwt->pageidx*PSYNC_FS_PAGE_SIZE+pwt->rsize-pageoff<ret)
ret=pwt->pageidx*PSYNC_FS_PAGE_SIZE+pwt->rsize-pageoff;
}
else
ret=pwt->rsize;
}
}
}
psync_list_for_each_element_call(&waiting, psync_page_waiter_t, listwaiter, psync_free_page_waiter);
}
return ret;
}
static void free_waiters(psync_list *waiters){
psync_list *l1, *l2;
psync_page_waiter_t *pwt;
psync_page_wait_t *pw;
psync_list_for_each_safe (l1, l2, waiters){
pwt=psync_list_element(l1, psync_page_waiter_t, listwaiter);
if (!pwt->ready){
psync_list_del(&pwt->listpage);
pw=pwt->waiting_for;
if (psync_list_isempty(&pw->waiters)){
psync_list_del(&pw->list);
psync_free(pw);
}
}
psync_free_page_waiter(pwt);
}
}
static int request_auth_page(psync_crypto_auth_page *ap, psync_request_t *rq, psync_list *waiting, psync_fileid_t fileid,
uint64_t hash, uint64_t aoffset, uint32_t asize){
uint64_t apageid;
char *pbuff;
psync_uint_t apoff, apsize;
psync_int_t rb;
apageid=aoffset/PSYNC_FS_PAGE_SIZE;
if (aoffset%PSYNC_FS_PAGE_SIZE==0){
rb=check_page_in_memory_by_hash(hash, apageid, (char *)ap->auth, asize, 0);
if (rb==-1)
rb=check_page_in_database_by_hash_and_cache(hash, apageid, (char *)ap->auth, asize, 0);
if (rb==-1)
ap->waiter=add_page_waiter(waiting, &rq->ranges, hash, apageid, fileid, (char *)ap->auth, 0, 0, asize);
else if (unlikely(rb!=asize)){
debug(D_WARNING, "expected auth sector size to be %u, got %u", (unsigned)asize, (unsigned)rb);
return -1;
}
}
else{
apoff=aoffset-apageid*PSYNC_FS_PAGE_SIZE;
apsize=PSYNC_FS_PAGE_SIZE-apoff;
if (apsize>asize)
apsize=asize;
rb=check_page_in_memory_by_hash(hash, apageid, (char *)ap->auth, apsize, apoff);
if (rb==-1)
rb=check_page_in_database_by_hash_and_cache(hash, apageid, (char *)ap->auth, apsize, apoff);
if (rb==-1)
ap->waiter=add_page_waiter(waiting, &rq->ranges, hash, apageid, fileid, (char *)ap->auth, 0, apoff, apsize);
else if (unlikely(rb!=apsize)){
debug(D_WARNING, "expected auth sector size to be %u, got %u", (unsigned)asize, (unsigned)rb);
return -1;
}
if (apsize<asize){
pbuff=((char *)ap->auth)+apsize;
apageid++;
apsize=asize-apsize;
assert(apsize<PSYNC_FS_PAGE_SIZE);
rb=check_page_in_memory_by_hash(hash, apageid, pbuff, apsize, 0);
if (rb==-1)
rb=check_page_in_database_by_hash_and_cache(hash, apageid, pbuff, apsize, 0);
if (rb==-1)
ap->waiter=add_page_waiter(waiting, &rq->ranges, hash, apageid, fileid, pbuff, 0, 0, apsize);
else if (unlikely(rb!=apsize)){
debug(D_WARNING, "expected auth sector size to be %u, got %u", (unsigned)asize, (unsigned)rb);
return -1;
}
}
}
return 0;
}
int psync_pagecache_read_unmodified_encrypted_locked(psync_openfile_t *of, char *buf, uint64_t size, uint64_t offset){
psync_crypto_offsets_t offsets;
uint64_t initialsize, hash, poffset, psize, first_page_id, aoffset, apageid, authupto;
psync_uint_t i, pageoff, pagecnt, apsize;
psync_int_t rb;
psync_request_t *rq;
psync_crypto_auth_page *ap;
psync_crypto_data_page *dp;
char *pbuff;
psync_interval_tree_t *intv;
psync_list auth_pages, waiting;
psync_fileid_t fileid;
uint32_t asize, aoff;
int ret, needkey;
initialsize=of->initialsize;
hash=of->hash;
fileid=of->remotefileid;
intv=psync_interval_tree_first_interval_containing_or_after(of->authenticatedints, offset);
if (!intv || intv->from>offset)
authupto=0;
else
authupto=intv->to;
if (of->encoder==PSYNC_CRYPTO_UNLOADED_SECTOR_ENCODER && initialsize && offset<initialsize){
needkey=1;
of->encoder=PSYNC_CRYPTO_LOADING_SECTOR_ENCODER;
}
else
needkey=0;
pthread_mutex_unlock(&of->mutex);
assert(PSYNC_CRYPTO_SECTOR_SIZE==PSYNC_FS_PAGE_SIZE);
psync_fs_crypto_offsets_by_plainsize(initialsize, &offsets);
if (offset>=initialsize)
return 0;
if (offset+size>initialsize)
size=initialsize-offset;
poffset=offset_round_down_to_page(offset);
pageoff=offset-poffset;
psize=size_round_up_to_page(size+pageoff);
pagecnt=psize/PSYNC_FS_PAGE_SIZE;
first_page_id=poffset/PSYNC_FS_PAGE_SIZE;
rq=psync_new(psync_request_t);
psync_list_init(&rq->ranges);
psync_list_init(&waiting);
psync_list_init(&auth_pages);
dp=psync_new_cnt(psync_crypto_data_page, pagecnt);
memset(dp, 0, sizeof(psync_crypto_data_page)*pagecnt);
ap=NULL;
lock_wait(hash);
for (i=0; i<pagecnt; i++){
if (i && (first_page_id+i)/PSYNC_CRYPTO_HASH_TREE_SECTORS==(first_page_id+i-1)/PSYNC_CRYPTO_HASH_TREE_SECTORS){
dp[i].authpage=ap;
continue;
}
psync_fs_crypto_get_auth_sector_off(first_page_id+i, 0, &offsets, &aoffset, &asize, &aoff);
ap=psync_new(psync_crypto_auth_page);
ap->waiter=NULL;
ap->parent=NULL;
ap->firstpageid=(first_page_id+i)/PSYNC_CRYPTO_HASH_TREE_SECTORS*PSYNC_CRYPTO_HASH_TREE_SECTORS;
ap->size=asize;
ap->idinparent=0;
ap->level=0;
psync_list_add_tail(&auth_pages, &ap->list);
dp[i].authpage=ap;
if (request_auth_page(ap, rq, &waiting, fileid, hash, aoffset, asize))
goto err0;
if (authupto<(first_page_id+i+1)*PSYNC_FS_PAGE_SIZE && offsets.needmasterauth){
psync_crypto_auth_page *lap, *cap;
psync_uint_t l;
lap=ap;
for (l=1; l<=offsets.treelevels; l++){
psync_fs_crypto_get_auth_sector_off(first_page_id+i, l, &offsets, &aoffset, &asize, &aoff);
cap=psync_new(psync_crypto_auth_page);
cap->waiter=NULL;
cap->parent=NULL;
cap->firstpageid=0;
cap->size=asize;
cap->idinparent=0;
cap->level=l;
lap->parent=cap;
lap->idinparent=aoff;
lap=cap;
psync_list_add_tail(&auth_pages, &cap->list);
if (request_auth_page(cap, rq, &waiting, fileid, hash, aoffset, asize))
goto err0;
}
}
}
for (i=0; i<pagecnt; i++){
pbuff=NULL;
if ((first_page_id+i)*PSYNC_FS_PAGE_SIZE+PSYNC_FS_PAGE_SIZE>initialsize){
apsize=initialsize-((first_page_id+i)*PSYNC_FS_PAGE_SIZE);
assert(apsize>0 && apsize<=PSYNC_FS_PAGE_SIZE);
}
else
apsize=PSYNC_FS_PAGE_SIZE;
if (i==0){
if (pageoff==0 && size>apsize)
pbuff=buf;
}
else if (i==pagecnt-1){
if (i*PSYNC_FS_PAGE_SIZE+apsize<=size+pageoff)
pbuff=buf+i*PSYNC_FS_PAGE_SIZE-pageoff;
}
else
pbuff=buf+i*PSYNC_FS_PAGE_SIZE-pageoff;
if (!pbuff){
pbuff=psync_new_cnt(char, apsize);
dp[i].freebuff=1;
}
dp[i].buff=pbuff;
dp[i].pagesize=apsize;
apageid=psync_fs_crypto_data_sectorid_by_sectorid(first_page_id+i);
rb=check_page_in_memory_by_hash(hash, apageid, pbuff, apsize, 0);
if (rb==-1)
rb=check_page_in_database_by_hash(hash, apageid, pbuff, apsize, 0);
if (rb==-1)
dp[i].waiter=add_page_waiter(&waiting, &rq->ranges, hash, apageid, fileid, pbuff, i, 0, apsize);
else if (unlikely(rb!=apsize))
goto err0;
}
unlock_wait(hash);
psync_pagecache_read_unmodified_readahead(of, poffset, psize, &rq->ranges, fileid, hash, initialsize, &offsets);
if (!psync_list_isempty(&rq->ranges) || needkey){
rq->of=of;
rq->fileid=fileid;
rq->hash=hash;
rq->needkey=needkey;
psync_fs_inc_of_refcnt_and_readers(of);
psync_run_thread1("crypto read unmodified", psync_pagecache_read_unmodified_thread, rq);
}
else
psync_free(rq);
ret=0;
if (needkey){
debug(D_NOTICE, "waiting for key to download");
psync_fs_lock_file(of);
while (of->encoder==PSYNC_CRYPTO_LOADING_SECTOR_ENCODER)
pthread_cond_wait(&enc_key_cond, &of->mutex);
if (of->encoder==PSYNC_CRYPTO_FAILED_SECTOR_ENCODER){
debug(D_NOTICE, "failed to download key");
ret=-EIO;
}
pthread_mutex_unlock(&of->mutex);
debug(D_NOTICE, "waited for key to download");
}
for (i=0; i<pagecnt; i++){
ap=dp[i].authpage;
if (ap->waiter){
wait_waiter(ap->waiter, hash, "auth");
if (!ret && ap->waiter->error)
ret=ap->waiter->error;
}
if (ap->parent && !ret){
#ifdef P_NO_CHECKSUM_CHECK
#if IS_DEBUG
debug(D_NOTICE, "NOT checking chain checksums for pages %lu-%lu tree level %d",
(unsigned long)ap->firstpageid, (unsigned long)ap->firstpageid+ap->size/PSYNC_CRYPTO_AUTH_SIZE, (int)offsets.treelevels);
psync_fs_lock_file(of);
if (likely(of->hash==hash))
psync_interval_tree_add(&of->authenticatedints, ap->firstpageid*PSYNC_FS_PAGE_SIZE, (ap->firstpageid+ap->size/PSYNC_CRYPTO_AUTH_SIZE)*PSYNC_FS_PAGE_SIZE);
pthread_mutex_unlock(&of->mutex);
#else
abort();
#endif
#else
psync_crypto_sector_auth_t sa;
psync_crypto_auth_page *p;
debug(D_NOTICE, "checking chain checksums for pages %lu-%lu tree level %d",
(unsigned long)ap->firstpageid, (unsigned long)ap->firstpageid+ap->size/PSYNC_CRYPTO_AUTH_SIZE, (int)offsets.treelevels);
psync_crypto_sign_auth_sector(of->encoder, (unsigned char *)ap->auth, ap->size, sa);
p=ap->parent;
ap->parent=NULL;
do {
if (p->waiter){
wait_waiter(p->waiter, hash, "chain auth");
if (!ret && p->waiter->error)
ret=p->waiter->error;
}
if (!ret && memcmp(sa, p->auth[ap->idinparent], sizeof(psync_crypto_sector_auth_t))){
debug(D_ERROR, "chain verification failed for sector %lu at level %u, idinparent=%u",
(unsigned long)(first_page_id+i), (unsigned)p->level, (unsigned)ap->idinparent);
ret=-EIO;
}
ap=p;
psync_crypto_sign_auth_sector(of->encoder, (unsigned char *)ap->auth, ap->size, sa);
p=ap->parent;
} while (p);
ap=dp[i].authpage;
if (!ret){
psync_fs_lock_file(of);
if (likely(of->hash==hash))
psync_interval_tree_add(&of->authenticatedints, ap->firstpageid*PSYNC_FS_PAGE_SIZE, (ap->firstpageid+ap->size/PSYNC_CRYPTO_AUTH_SIZE)*PSYNC_FS_PAGE_SIZE);
pthread_mutex_unlock(&of->mutex);
}
#endif
}
if (dp[i].waiter){
wait_waiter(dp[i].waiter, hash, "data");
if (!ret && dp[i].waiter->error)
ret=dp[i].waiter->error;
}
if (!ret){
apageid=first_page_id+i-ap->firstpageid;
assert(apageid>=0 && apageid<PSYNC_CRYPTO_HASH_TREE_SECTORS);
if (psync_crypto_aes256_decode_sector(of->encoder, (unsigned char *)dp[i].buff, dp[i].pagesize, (unsigned char *)dp[i].buff,
ap->auth[apageid], first_page_id+i)){
debug(D_ERROR, "decoding of page %lu of file %s failed pagesize=%u, requested offset=%lu, requested size=%lu",
(unsigned long)(first_page_id+i), of->currentname, (unsigned)dp[i].pagesize, (unsigned long)offset, (unsigned long)size);
ret=-EIO;
}
else if (dp[i].freebuff){
uint64_t copysize;
psync_uint_t copyoff;
if (i==0){
copyoff=pageoff;
if (size>PSYNC_FS_PAGE_SIZE-copyoff)
copysize=PSYNC_FS_PAGE_SIZE-copyoff;
else
copysize=size;
pbuff=buf;
}
else{
assert(i==pagecnt-1);
copyoff=0;
copysize=(size+pageoff)&(PSYNC_FS_PAGE_SIZE-1);
if (!copysize)
copysize=PSYNC_FS_PAGE_SIZE;
pbuff=buf+i*PSYNC_FS_PAGE_SIZE-pageoff;
}
memcpy(pbuff, dp[i].buff+copyoff, copysize);
}
}
}
if (!ret)
ret=size;
ret0:
psync_list_for_each_element_call(&waiting, psync_page_waiter_t, listwaiter, psync_free);
psync_list_for_each_element_call(&auth_pages, psync_crypto_auth_page, list, psync_free);
for (i=0; i<pagecnt; i++)
if (dp[i].freebuff)
psync_free(dp[i].buff);
psync_free(dp);
return ret;
err0:
free_waiters(&waiting);
unlock_wait(hash);
psync_pagecache_free_request(rq);
ret=-EIO;
goto ret0;
}
int psync_pagecache_readv_locked(psync_openfile_t *of, psync_pagecache_read_range *ranges, int cnt){
uint64_t initialsize, hash, poffset, psize, first_page_id;
psync_uint_t pageoff, copyoff, copysize, pagecnt, j;
psync_int_t rb;
psync_fileid_t fileid;
psync_request_t *rq;
psync_page_waiter_t *pwt;
char *pbuff;
psync_list waiting;
int i, ret, needkey;
initialsize=of->initialsize;
hash=of->hash;
fileid=of->remotefileid;
if (of->encrypted && of->encoder==PSYNC_CRYPTO_UNLOADED_SECTOR_ENCODER && initialsize){
needkey=1;
of->encoder=PSYNC_CRYPTO_LOADING_SECTOR_ENCODER;
}
else
needkey=0;
pthread_mutex_unlock(&of->mutex);
rq=psync_new(psync_request_t);
psync_list_init(&rq->ranges);
psync_list_init(&waiting);
for (i=0; i<cnt; i++){
assert(ranges[i].offset+ranges[i].size<=of->encrypted?psync_fs_crypto_crypto_size(initialsize):initialsize);
poffset=offset_round_down_to_page(ranges[i].offset);
pageoff=ranges[i].offset-poffset;
psize=size_round_up_to_page(ranges[i].size+pageoff);
pagecnt=psize/PSYNC_FS_PAGE_SIZE;
first_page_id=poffset/PSYNC_FS_PAGE_SIZE;
for (j=0; j<pagecnt; j++){
if (j==0){
copyoff=pageoff;
if (ranges[i].size>PSYNC_FS_PAGE_SIZE-copyoff)
copysize=PSYNC_FS_PAGE_SIZE-copyoff;
else
copysize=ranges[i].size;
pbuff=ranges[i].buf;
}
else if (j==pagecnt-1){
copyoff=0;
copysize=(ranges[i].size+pageoff)&(PSYNC_FS_PAGE_SIZE-1);
if (!copysize)
copysize=PSYNC_FS_PAGE_SIZE;
pbuff=ranges[i].buf+j*PSYNC_FS_PAGE_SIZE-pageoff;
}
else{
copyoff=0;
copysize=PSYNC_FS_PAGE_SIZE;
pbuff=ranges[i].buf+j*PSYNC_FS_PAGE_SIZE-pageoff;
}
rb=check_page_in_memory_by_hash(hash, first_page_id+j, pbuff, copysize, copyoff);
if (rb==-1)
rb=check_page_in_database_by_hash(hash, first_page_id+j, pbuff, copysize, copyoff);
if (rb!=-1){
if (likely(rb==copysize))
continue;
else{
lock_wait(hash);
free_waiters(&waiting);
unlock_wait(hash);
}
}
lock_wait(hash);
add_page_waiter(&waiting, &rq->ranges, hash, first_page_id+j, fileid, pbuff, j, copyoff, copysize);
unlock_wait(hash);
}
}
if (psync_list_isempty(&rq->ranges) && !needkey){
if (psync_list_isempty(&waiting)){
psync_free(rq);
return 0;
}
}
else{
rq->of=of;
rq->fileid=fileid;
rq->hash=hash;
rq->needkey=needkey;
psync_fs_inc_of_refcnt_and_readers(of);
psync_run_thread1("readv unmodified", psync_pagecache_read_unmodified_thread, rq);
}
ret=0;
if (needkey){
debug(D_NOTICE, "waiting for key to download");
psync_fs_lock_file(of);
while (of->encoder==PSYNC_CRYPTO_LOADING_SECTOR_ENCODER)
pthread_cond_wait(&enc_key_cond, &of->mutex);
if (of->encoder==PSYNC_CRYPTO_FAILED_SECTOR_ENCODER){
debug(D_NOTICE, "failed to download key");
ret=-1;
}
pthread_mutex_unlock(&of->mutex);
debug(D_NOTICE, "waited for key to download");
}
lock_wait(hash);
psync_list_for_each_element(pwt, &waiting, psync_page_waiter_t, listwaiter){
while (!pwt->ready){
debug(D_NOTICE, "waiting for page #%lu to be read", (unsigned long)pwt->waiting_for->pageid);
pthread_cond_wait(&pwt->cond, &wait_page_mutex);
debug(D_NOTICE, "waited for page"); // not safe to use pwt->waiting_for here
}
if (pwt->error || pwt->rsize<pwt->size)
ret=-1;
}
unlock_wait(hash);
psync_list_for_each_element_call(&waiting, psync_page_waiter_t, listwaiter, psync_free_page_waiter);
return ret;
}
static void psync_pagecache_add_page_if_not_exists(psync_cache_page_t *page, uint64_t hash, uint64_t pageid){
psync_cache_page_t *pg;
psync_page_wait_t *pw;
psync_uint_t h1, h2;
int hasit;
hasit=0;
if (has_page_in_db(hash, pageid)){
psync_pagecache_return_free_page(page);
return;
}
h1=pagehash_by_hash_and_pageid(hash, pageid);
h2=waiterhash_by_hash_and_pageid(hash, pageid);
lock_wait(hash);
pthread_mutex_lock(&cache_mutex);
psync_list_for_each_element(pg, &cache_hash[h1], psync_cache_page_t, list)
if (pg->type==PAGE_TYPE_READ && pg->hash==hash && pg->pageid==pageid){
hasit=1;
break;
}
if (!hasit)
psync_list_for_each_element(pw, &wait_page_hash[h2], psync_page_wait_t, list)
if (pw->hash==hash && pw->pageid==pageid){
hasit=1;
break;
}
if (hasit)
psync_pagecache_return_free_page_locked(page);
else{
psync_list_add_tail(&cache_hash[h1], &page->list);
cache_pages_in_hash++;
}
pthread_mutex_unlock(&cache_mutex);
unlock_wait(hash);
}
static void psync_check_clean_running(){
if (unlikely(clean_cache_in_progress)){
if (pthread_mutex_trylock(&clean_cache_mutex)){
debug(D_NOTICE, "waiting for cache clean to exit");
pthread_mutex_lock(&clean_cache_mutex);
pthread_mutex_unlock(&clean_cache_mutex);
debug(D_NOTICE, "waited for cache clean to exit");
}
else
pthread_mutex_unlock(&clean_cache_mutex);
}
}
static void psync_pagecache_new_upload_to_cache(uint64_t taskid, uint64_t hash, int pause){
char *filename;
psync_cache_page_t *page;
uint64_t pageid;
ssize_t rd;
time_t tm;
psync_file_t fd;
char fileidhex[sizeof(psync_fsfileid_t)*2+2];
psync_binhex(fileidhex, &taskid, sizeof(psync_fsfileid_t));
fileidhex[sizeof(psync_fsfileid_t)]='d';
fileidhex[sizeof(psync_fsfileid_t)+1]=0;
tm=psync_timer_time();
filename=psync_strcat(psync_setting_get_string(_PS(fscachepath)), PSYNC_DIRECTORY_SEPARATOR, fileidhex, NULL);
fd=psync_file_open(filename, P_O_RDONLY, 0);
if (fd==INVALID_HANDLE_VALUE){
debug(D_ERROR, "could not open cache file %s for taskid %lu, skipping", filename, (unsigned long)taskid);
psync_file_delete(filename);
psync_free(filename);
return;
}
debug(D_NOTICE, "adding file %s to cache for hash %lu (%ld) size %ld", filename, (unsigned long)hash, (long)hash, (long)psync_file_size(fd));
pageid=0;
while (1){
page=psync_pagecache_get_free_page(1);
rd=psync_file_read(fd, page->page, PSYNC_FS_PAGE_SIZE);
if (rd<=0){
psync_pagecache_return_free_page(page);
break;
}
page->hash=hash;
page->pageid=pageid;
page->lastuse=tm;
page->size=rd;
page->usecnt=1;
page->crc=psync_crc32c(PSYNC_CRC_INITIAL, page->page, rd);
page->type=PAGE_TYPE_READ;
psync_pagecache_add_page_if_not_exists(page, hash, pageid);
if (rd<PSYNC_FS_PAGE_SIZE)
break;
pageid++;
if (pause && pageid%64==0){
psync_check_clean_running();
psync_milisleep(10);
}
}
psync_file_close(fd);
psync_file_delete(filename);
debug(D_NOTICE, "finished adding file %s to cache for hash %lu (%ld)", filename, (unsigned long)hash, (long)hash);
psync_free(filename);
}
static void switch_pageids(uint64_t hash, uint64_t oldhash, uint64_t *pageids, psync_uint_t pageidcnt){
psync_sql_res *res;
psync_uint_t i;
res=psync_sql_prep_statement("UPDATE OR IGNORE pagecache SET hash=?, lastuse=? WHERE hash=? AND type=? AND pageid=?");
psync_sql_start_transaction();
psync_sql_bind_uint(res, 1, hash);
psync_sql_bind_uint(res, 2, psync_timer_time());
psync_sql_bind_uint(res, 3, oldhash);
psync_sql_bind_uint(res, 4, PAGE_TYPE_READ);
for (i=0; i<pageidcnt; i++){
psync_sql_bind_uint(res, 5, pageids[i]);
psync_sql_run(res);
}
psync_sql_commit_transaction();
psync_sql_free_result(res);
}
static void psync_pagecache_modify_to_cache(uint64_t taskid, uint64_t hash, uint64_t oldhash){
uint64_t pageids[64];
char *filename, *indexname;
const char *cachepath;
psync_cache_page_t *page;
psync_interval_tree_t *tree, *interval;
uint64_t pageid, off, roff, rdoff, rdlen;
int64_t fs;
ssize_t rd;
psync_int_t pdb;
psync_uint_t pageidcnt;
time_t tm;
psync_file_t fd;
char fileidhex[sizeof(psync_fsfileid_t)*2+2];
int ret;
pageidcnt=0;
psync_binhex(fileidhex, &taskid, sizeof(psync_fsfileid_t));
fileidhex[sizeof(psync_fsfileid_t)]='d';
fileidhex[sizeof(psync_fsfileid_t)+1]=0;
tm=psync_timer_time();
cachepath=psync_setting_get_string(_PS(fscachepath));
filename=psync_strcat(cachepath, PSYNC_DIRECTORY_SEPARATOR, fileidhex, NULL);
fileidhex[sizeof(psync_fsfileid_t)]='i';
indexname=psync_strcat(cachepath, PSYNC_DIRECTORY_SEPARATOR, fileidhex, NULL);
fd=psync_file_open(indexname, P_O_RDONLY, 0);
if (unlikely(fd==INVALID_HANDLE_VALUE)){
debug(D_ERROR, "could not open index of cache file %s for taskid %lu, skipping", indexname, (unsigned long)taskid);
psync_file_delete(filename);
psync_file_delete(indexname);
psync_free(filename);
psync_free(indexname);
return;
}
tree=NULL;
if (unlikely_log((fs=psync_file_size(fd))==-1 || psync_fs_load_interval_tree(fd, fs, &tree)==-1))
goto err2;
psync_file_close(fd);
fd=psync_file_open(filename, P_O_RDONLY, 0);
if (unlikely(fd==INVALID_HANDLE_VALUE)){
debug(D_ERROR, "could not open cache file %s for taskid %lu, skipping", filename, (unsigned long)taskid);
goto err1;
}
fs=psync_file_size(fd);
if (unlikely_log(fs==-1))
goto err2;
debug(D_NOTICE, "adding blocks of file %s to cache for hash %lu (%ld), old hash %lu (%ld) size %ld",
filename, (unsigned long)hash, (long)hash, (unsigned long)oldhash, (long)oldhash, (long)fs);
interval=psync_interval_tree_get_first(tree);
for (off=0; off<fs; off+=PSYNC_FS_PAGE_SIZE){
pageid=off/PSYNC_FS_PAGE_SIZE;
while (interval && interval->to<=off)
interval=psync_interval_tree_get_next(interval);
if (!interval || interval->from>=off+PSYNC_FS_PAGE_SIZE){ // full old page
if (unlikely(off+PSYNC_FS_PAGE_SIZE>fs)){ // last page, if it is equal we are ok
page=psync_pagecache_get_free_page(1);
pdb=check_page_in_memory_by_hash(oldhash, pageid, page->page, fs-off, 0);
if (pdb==-1)
pdb=check_page_in_database_by_hash(oldhash, pageid, page->page, fs-off, 0);
if (pdb==-1){
psync_pagecache_return_free_page(page);
break;
}
page->hash=hash;
page->pageid=pageid;
page->lastuse=tm;
page->size=pdb;
page->usecnt=1;
page->crc=psync_crc32c(PSYNC_CRC_INITIAL, page->page, pdb);
page->type=PAGE_TYPE_READ;
psync_pagecache_add_page_if_not_exists(page, hash, pageid);
break;
}
if (!switch_memory_page_to_hash(oldhash, hash, pageid)){
pageids[pageidcnt]=pageid;
if (++pageidcnt==ARRAY_SIZE(pageids)){
pageidcnt=0;
switch_pageids(hash, oldhash, pageids, ARRAY_SIZE(pageids));
psync_milisleep(10);
psync_check_clean_running();
}
}
}
else if (interval->from<=off && (interval->to>=off+PSYNC_FS_PAGE_SIZE || interval->to>=fs)){ // full new page
page=psync_pagecache_get_free_page(1);
if (off+PSYNC_FS_PAGE_SIZE>fs)
rd=fs-off;
else
rd=PSYNC_FS_PAGE_SIZE;
rd=psync_file_pread(fd, page->page, rd, off);
if (rd<PSYNC_FS_PAGE_SIZE && off+rd!=fs){
psync_pagecache_return_free_page(page);
break;
}
page->hash=hash;
page->pageid=pageid;
page->lastuse=tm;
page->size=rd;
page->usecnt=1;
page->crc=psync_crc32c(PSYNC_CRC_INITIAL, page->page, rd);
page->type=PAGE_TYPE_READ;
// debug(D_NOTICE, "new page %lu crc %lu size %lu", (unsigned long)pageid, (unsigned long)page->crc, (unsigned long)rd);
psync_pagecache_add_page_if_not_exists(page, hash, pageid);
if (pageid%64==0){
psync_check_clean_running();
psync_milisleep(10);
}
}
else{ // page with both old and new fragments
// we covered full new page and full old page cases, so this interval either ends or starts inside current page
assert((interval->to>off && interval->to<=off+PSYNC_FS_PAGE_SIZE) || (interval->from>=off && interval->from<off+PSYNC_FS_PAGE_SIZE));
page=psync_pagecache_get_free_page(1);
pdb=check_page_in_memory_by_hash(oldhash, pageid, page->page, PSYNC_FS_PAGE_SIZE, 0);
if (pdb==-1)
pdb=check_page_in_database_by_hash(oldhash, pageid, page->page, PSYNC_FS_PAGE_SIZE, 0);
if (pdb==-1){
psync_pagecache_return_free_page(page);
continue;
}
ret=0;
while (1){
if (interval->from>off){
roff=interval->from-off;
rdoff=interval->from;
}
else{
roff=0;
rdoff=off;
}
if (interval->to<off+PSYNC_FS_PAGE_SIZE)
rdlen=interval->to-rdoff;
else
rdlen=PSYNC_FS_PAGE_SIZE-roff;
assert(roff+rdlen<=PSYNC_FS_PAGE_SIZE);
// debug(D_NOTICE, "ifrom=%lu ito=%lu roff=%lu roff=%lu rdlen=%lu", interval->from, interval->to, rdoff, roff, rdlen);
rd=psync_file_pread(fd, page->page+roff, rdlen, rdoff);
if (rd!=rdlen){
if (rd>=0 && rdoff+rd>=fs) {
if (roff+rd>pdb)
pdb=roff+rd;
break;
}
ret=-1;
break;
}
if (roff+rd>pdb)
pdb=roff+rd;
if (interval->to>off+PSYNC_FS_PAGE_SIZE)
break;
interval=psync_interval_tree_get_next(interval);
if (!interval || interval->from>=off+PSYNC_FS_PAGE_SIZE)
break;
}
if (unlikely_log(ret==-1)){
psync_pagecache_return_free_page(page);
continue;
}
if (pdb+off>fs){
debug(D_NOTICE, "%lu+%lu>%lu", (unsigned long)pdb, (unsigned long)off, (unsigned long)fs);
assert(fs>off);
pdb=fs-off;
}
page->hash=hash;
page->pageid=pageid;
page->lastuse=tm;
page->size=pdb;
page->usecnt=1;
page->crc=psync_crc32c(PSYNC_CRC_INITIAL, page->page, pdb);
page->type=PAGE_TYPE_READ;
// debug(D_NOTICE, "combined page %lu crc %lu size %lu", (unsigned long)pageid, (unsigned long)page->crc, (unsigned long)pdb);
psync_pagecache_add_page_if_not_exists(page, hash, pageid);
if (pageid%64==0){
psync_check_clean_running();
psync_milisleep(10);
}
}
}
err2:
if (pageidcnt)
switch_pageids(hash, oldhash, pageids, pageidcnt);
psync_file_close(fd);
err1:
psync_interval_tree_free(tree);
psync_file_delete(filename);
psync_free(filename);
psync_file_delete(indexname);
psync_free(indexname);
}
static void psync_pagecache_check_free_space(){
const char *cachepath;
uint64_t minlocal;
int64_t freespc;
cachepath=psync_setting_get_string(_PS(fscachepath));
minlocal=psync_setting_get_uint(_PS(minlocalfreespace));
freespc=psync_get_free_space_by_path(cachepath);
if (unlikely(freespc==-1)){
debug(D_WARNING, "could not get free space of path %s", cachepath);
return;
}
if (freespc>=minlocal)
psync_set_local_full(0);
else{
debug(D_WARNING, "local disk holding %s is full", cachepath);
psync_set_local_full(1);
}
}
static void psync_pagecache_upload_to_cache(){
psync_sql_res *res;
psync_uint_row row;
uint64_t id, type, taskid, hash, oldhash;
uint32_t wake;
while (1){
res=psync_sql_query("SELECT id, type, taskid, hash, oldhash FROM pagecachetask ORDER BY id LIMIT 1");
row=psync_sql_fetch_rowint(res);
if (!row){
upload_to_cache_thread_run=0;
psync_sql_free_result(res);
break;
}
id=row[0];
type=row[1];
taskid=row[2];
hash=row[3];
oldhash=row[4];
psync_sql_free_result(res);
if (type==PAGE_TASK_TYPE_CREAT)
psync_pagecache_new_upload_to_cache(taskid, hash, 1);
else if (type==PAGE_TASK_TYPE_MODIFY)
psync_pagecache_modify_to_cache(taskid, hash, oldhash);
psync_sql_start_transaction();
res=psync_sql_prep_statement("DELETE FROM fstaskdepend WHERE dependfstaskid=?");
psync_sql_bind_uint(res, 1, taskid);
psync_sql_run_free(res);
wake=psync_sql_affected_rows();
res=psync_sql_prep_statement("DELETE FROM fstask WHERE id=?");
psync_sql_bind_uint(res, 1, taskid);
psync_sql_run_free(res);
if (IS_DEBUG) {
if (psync_sql_affected_rows())
debug(D_NOTICE, "deleted taskid %lu from fstask", (unsigned long)taskid);
else
debug(D_NOTICE, "no affected rows for deletion of taskid %lu from fstask", (unsigned long)taskid);
}
res=psync_sql_prep_statement("DELETE FROM pagecachetask WHERE id=?");
psync_sql_bind_uint(res, 1, id);
psync_sql_run_free(res);
psync_sql_commit_transaction();
if (wake)
psync_fsupload_wake();
psync_pagecache_check_free_space();
}
}
static void psync_pagecache_add_task(uint32_t type, uint64_t taskid, uint64_t hash, uint64_t oldhash){
psync_sql_res *res;
int run;
run=0;
res=psync_sql_prep_statement("INSERT INTO pagecachetask (type, taskid, hash, oldhash) VALUES (?, ?, ?, ?)");
psync_sql_bind_uint(res, 1, type);
psync_sql_bind_uint(res, 2, taskid);
psync_sql_bind_uint(res, 3, hash);
psync_sql_bind_uint(res, 4, oldhash);
if (!upload_to_cache_thread_run){
upload_to_cache_thread_run=1;
run=1;
}
psync_sql_run_free(res);
if (run)
psync_run_thread("upload to cache", psync_pagecache_upload_to_cache);
}
void psync_pagecache_creat_to_pagecache(uint64_t taskid, uint64_t hash, int onthisthread){
if (onthisthread)
psync_pagecache_new_upload_to_cache(taskid, hash, 0);
else
psync_pagecache_add_task(PAGE_TASK_TYPE_CREAT, taskid, hash, 0);
}
void psync_pagecache_modify_to_pagecache(uint64_t taskid, uint64_t hash, uint64_t oldhash){
psync_pagecache_add_task(PAGE_TASK_TYPE_MODIFY, taskid, hash, oldhash);
}
int psync_pagecache_have_all_pages_in_cache(uint64_t hash, uint64_t size){
unsigned char *db;
uint32_t i, pagecnt;
pagecnt=(size+PSYNC_FS_PAGE_SIZE-1)/PSYNC_FS_PAGE_SIZE;
db=has_pages_in_db(hash, 0, pagecnt, 0);
for (i=0; i<pagecnt; i++)
if (!db[i] && !has_page_in_cache_by_hash(hash, i))
break;
psync_free(db);
return i==pagecnt;
}
int psync_pagecache_copy_all_pages_from_cache_to_file_locked(psync_openfile_t *of, uint64_t hash, uint64_t size){
char buff[PSYNC_FS_PAGE_SIZE];
uint64_t i, pagecnt;
psync_int_t rb;
pagecnt=(size+PSYNC_FS_PAGE_SIZE-1)/PSYNC_FS_PAGE_SIZE;
for (i=0; i<pagecnt; i++){
rb=check_page_in_memory_by_hash(hash, i, buff, PSYNC_FS_PAGE_SIZE, 0);
if (rb==-1){
rb=check_page_in_database_by_hash(hash, i, buff, PSYNC_FS_PAGE_SIZE, 0);
if (rb==-1)
return -1;
}
assertw(rb==PSYNC_FS_PAGE_SIZE || i*PSYNC_FS_PAGE_SIZE+rb==size);
if (psync_file_pwrite(of->datafile, buff, rb, i*PSYNC_FS_PAGE_SIZE)!=rb)
return -1;
}
if (unlikely_log(psync_file_sync(of->datafile)))
return -1;
else{
debug(D_NOTICE, "copied %lu bytes to data file of %s from cache", (unsigned long)size, of->currentname);
return 0;
}
}
int psync_pagecache_lock_pages_in_cache(){
if (pthread_mutex_trylock(&clean_cache_mutex))
return -1;
clean_cache_stoppers++;
pthread_mutex_unlock(&clean_cache_mutex);
return 0;
}
void psync_pagecache_unlock_pages_from_cache(){
pthread_mutex_lock(&clean_cache_mutex);
if (--clean_cache_stoppers==0 && clean_cache_waiters)
pthread_cond_broadcast(&clean_cache_cond);
pthread_mutex_unlock(&clean_cache_mutex);
}
void psync_pagecache_resize_cache(){
pthread_mutex_lock(&flush_cache_mutex);
db_cache_in_pages=psync_setting_get_uint(_PS(fscachesize))/PSYNC_FS_PAGE_SIZE;
db_cache_max_page=psync_sql_cellint("SELECT MAX(id) FROM pagecache", 0);
if (db_cache_max_page>db_cache_in_pages){
psync_sql_res *res;
psync_stat_t st;
res=psync_sql_prep_statement("DELETE FROM pagecache WHERE id>?");
psync_sql_bind_uint(res, 1, db_cache_in_pages);
psync_sql_run_free(res);
db_cache_max_page=db_cache_in_pages;
if (!psync_fstat(readcache, &st) && psync_stat_size(&st)>db_cache_in_pages*PSYNC_FS_PAGE_SIZE){
if (likely_log(psync_file_seek(readcache, db_cache_in_pages*PSYNC_FS_PAGE_SIZE, P_SEEK_SET)!=-1)){
assertw(psync_file_truncate(readcache)==0);
debug(D_NOTICE, "shrunk cache to %lu pages (%lu bytes)", (unsigned long)db_cache_in_pages, (unsigned long)db_cache_in_pages*PSYNC_FS_PAGE_SIZE);
}
}
}
pthread_mutex_unlock(&flush_cache_mutex);
}
static int psync_pagecache_free_page_from_read_cache(){
psync_stat_t st;
uint64_t sizeinpages;
psync_sql_res *res;
psync_cache_page_t *page;
psync_uint_row row;
int ret;
ret=-1;
pthread_mutex_lock(&flush_cache_mutex);
do {
if (psync_fstat(readcache, &st)){
debug(D_NOTICE, "stat of read cache file failed");
break;
}
if (psync_stat_size(&st)<PSYNC_FS_PAGE_SIZE*2){
debug(D_NOTICE, "read cache is already zero");
break;
}
sizeinpages=psync_stat_size(&st)/PSYNC_FS_PAGE_SIZE-1;
if (unlikely(db_cache_max_page>sizeinpages)){
debug(D_NOTICE, "there are %lu unallocated pages in db, deleting", (unsigned long)(db_cache_max_page-sizeinpages));
res=psync_sql_prep_statement("DELETE FROM pagecache WHERE id>?");
psync_sql_bind_uint(res, 1, sizeinpages);
psync_sql_run_free(res);
db_cache_max_page=sizeinpages;
}
else if (unlikely_log(db_cache_max_page<sizeinpages))
sizeinpages=db_cache_max_page;
page=psync_pagecache_get_free_page_if_available();
if (unlikely(!page)){
debug(D_NOTICE, "no free pages, skipping");
break;
}
if (psync_file_pread(readcache, page->page, PSYNC_FS_PAGE_SIZE, sizeinpages*PSYNC_FS_PAGE_SIZE)!=PSYNC_FS_PAGE_SIZE){
psync_pagecache_return_free_page(page);
debug(D_NOTICE, "read from read cache failed");
break;
}
res=psync_sql_query_rdlock("SELECT type, hash, pageid, lastuse, usecnt, size, crc FROM pagecache WHERE id=?");
psync_sql_bind_uint(res, 1, sizeinpages);
row=psync_sql_fetch_rowint(res);
if (!row || row[0]==PAGE_TYPE_FREE){
psync_sql_free_result(res);
psync_pagecache_return_free_page(page);
}
else{
page->hash=row[1];
page->pageid=row[2];
page->lastuse=row[3];
page->size=row[5];
page->usecnt=row[4];
page->crc=row[6];
psync_sql_free_result(res);
if (unlikely(psync_crc32c(PSYNC_CRC_INITIAL, page->page, page->size)!=page->crc)){
debug(D_WARNING, "page CRC check failed, dropping page, db CRC %u calculated CRC %u",
(unsigned)page->crc, (unsigned)psync_crc32c(PSYNC_CRC_INITIAL, page->page, page->size));
psync_pagecache_return_free_page(page);
}
else{
page->type=PAGE_TYPE_READ;
psync_pagecache_add_page_if_not_exists(page, page->hash, page->pageid);
}
}
db_cache_max_page=sizeinpages-1;
res=psync_sql_prep_statement("DELETE FROM pagecache WHERE id>?");
psync_sql_bind_uint(res, 1, db_cache_max_page);
psync_sql_run_free(res);
if (psync_file_seek(readcache, sizeinpages*PSYNC_FS_PAGE_SIZE, P_SEEK_SET)!=-1 && psync_file_truncate(readcache)==0)
ret=0;
else
debug(D_NOTICE, "failed to truncate down read cache");
} while (0);
pthread_mutex_unlock(&flush_cache_mutex);
return ret;
}
uint64_t psync_pagecache_free_from_read_cache(uint64_t size){
uint64_t i;
size=size_round_up_to_page(size)/PSYNC_FS_PAGE_SIZE;
for (i=0; i<size; i++)
if (psync_pagecache_free_page_from_read_cache()){
debug(D_WARNING, "failed to free page from read cache");
break;
}
debug(D_NOTICE, "freed %lu pages from read cache", (unsigned long)i);
return i*PSYNC_FS_PAGE_SIZE;
}
void psync_pagecache_init(){
uint64_t i;
char *page_data, *cache_file;
const char *cache_dir;
psync_sql_res *res;
psync_cache_page_t *page;
psync_stat_t st;
for (i=0; i<CACHE_HASH; i++)
psync_list_init(&cache_hash[i]);
for (i=0; i<PAGE_WAITER_HASH; i++)
psync_list_init(&wait_page_hash[i]);
pthread_mutex_init(&wait_page_mutex, NULL);
psync_list_init(&free_pages);
memset(cachepages_to_update, 0, sizeof(cachepages_to_update));
pages_base=(char *)psync_mmap_anon_safe(CACHE_PAGES*(PSYNC_FS_PAGE_SIZE+sizeof(psync_cache_page_t)));
page_data=pages_base;
page=(psync_cache_page_t *)(page_data+CACHE_PAGES*PSYNC_FS_PAGE_SIZE);
cache_pages_free=CACHE_PAGES;
for (i=0; i<CACHE_PAGES; i++){
page->page=page_data;
psync_list_add_tail(&free_pages, &page->list);
page_data+=PSYNC_FS_PAGE_SIZE;
page++;
}
cache_dir=psync_setting_get_string(_PS(fscachepath));
if (psync_stat(cache_dir, &st))
psync_mkdir(cache_dir);
cache_file=psync_strcat(cache_dir, PSYNC_DIRECTORY_SEPARATOR, PSYNC_DEFAULT_READ_CACHE_FILE, NULL);
if (psync_stat(cache_file, &st))
psync_sql_statement("DELETE FROM pagecache");
else{
res=psync_sql_prep_statement("DELETE FROM pagecache WHERE id>?");
psync_sql_bind_uint(res, 1, psync_stat_size(&st)/PSYNC_FS_PAGE_SIZE);
psync_sql_run_free(res);
}
db_cache_in_pages=psync_setting_get_uint(_PS(fscachesize))/PSYNC_FS_PAGE_SIZE;
db_cache_max_page=psync_sql_cellint("SELECT MAX(id) FROM pagecache", 0);
free_db_pages=psync_sql_cellint("SELECT COUNT(*) FROM pagecache WHERE type="NTO_STR(PAGE_TYPE_FREE), 0);
if (db_cache_max_page<db_cache_in_pages && free_db_pages<CACHE_PAGES*2){
i=0;
psync_sql_start_transaction();
res=psync_sql_prep_statement("INSERT INTO pagecache (type) VALUES ("NTO_STR(PAGE_TYPE_FREE)")");
while (db_cache_max_page+i<db_cache_in_pages && i<CACHE_PAGES*2){
psync_sql_run(res);
i++;
}
psync_sql_free_result(res);
psync_sql_commit_transaction();
free_db_pages+=i;
db_cache_max_page+=i;
debug(D_NOTICE, "inserted %lu new free pages to database, db_cache_in_pages=%lu, db_cache_max_page=%lu",
(unsigned long)i, (unsigned long)db_cache_in_pages, (unsigned long)db_cache_max_page);
}
readcache=psync_file_open(cache_file, P_O_RDWR, P_O_CREAT);
psync_free(cache_file);
if (likely_log(psync_file_seek(readcache, db_cache_max_page*PSYNC_FS_PAGE_SIZE, P_SEEK_SET)!=-1))
assertw(psync_file_truncate(readcache)==0);
if (db_cache_max_page>db_cache_in_pages)
psync_pagecache_resize_cache();
pthread_mutex_lock(&flush_cache_mutex);
check_disk_full();
pthread_mutex_unlock(&flush_cache_mutex);
psync_sql_lock();
if (psync_sql_cellint("SELECT COUNT(*) FROM pagecachetask", 0)){
psync_run_thread("upload to cache", psync_pagecache_upload_to_cache);
upload_to_cache_thread_run=1;
}
psync_sql_unlock();
psync_timer_register(psync_pagecache_flush_timer, PSYNC_FS_DISK_FLUSH_SEC, NULL);
}
void clean_cache_del(void *delcache, psync_pstat *st){
int ret;
if (!psync_stat_isfolder(&st->stat) && (delcache || psync_filename_cmp(st->name, PSYNC_DEFAULT_READ_CACHE_FILE))){
ret=psync_file_delete(st->path);
debug(D_NOTICE, "delete of %s=%d", st->path, ret);
}
}
void psync_pagecache_clean_cache(){
const char *cache_dir;
cache_dir=psync_setting_get_string(_PS(fscachepath));
if (readcache!=INVALID_HANDLE_VALUE) {
psync_file_close(readcache);
readcache=INVALID_HANDLE_VALUE;
}
psync_list_dir(cache_dir, clean_cache_del, (void *)1);
}
void psync_pagecache_reopen_read_cache(){
char *cache_file;
const char *cache_dir;
psync_stat_t st;
cache_dir=psync_setting_get_string(_PS(fscachepath));
if (psync_stat(cache_dir, &st))
psync_mkdir(cache_dir);
cache_file=psync_strcat(cache_dir, PSYNC_DIRECTORY_SEPARATOR, PSYNC_DEFAULT_READ_CACHE_FILE, NULL);
readcache=psync_file_open(cache_file, P_O_RDWR, P_O_CREAT);
psync_free(cache_file);
}
void psync_pagecache_clean_read_cache(){
uint32_t i, cnt;
psync_sql_res *res;
debug(D_NOTICE, "start");
pthread_mutex_lock(&clean_cache_mutex);
pthread_mutex_lock(&flush_cache_mutex);
psync_sql_start_transaction();
debug(D_NOTICE, "aquired locks");
db_cache_in_pages=psync_setting_get_uint(_PS(fscachesize))/PSYNC_FS_PAGE_SIZE;
if (db_cache_in_pages<2*CACHE_PAGES)
cnt=db_cache_in_pages;
else
cnt=2*CACHE_PAGES;
psync_file_seek(readcache, cnt*PSYNC_FS_PAGE_SIZE, P_SEEK_SET);
assertw(psync_file_truncate(readcache)==0);
debug(D_NOTICE, "truncated cache file");
res=psync_sql_prep_statement("DELETE FROM pagecache");
psync_sql_run_free(res);
debug(D_NOTICE, "deleted entries from pagecache");
res=psync_sql_prep_statement("INSERT INTO pagecache (type) VALUES ("NTO_STR(PAGE_TYPE_FREE)")");
for (i=0; i<cnt; i++)
psync_sql_run(res);
psync_sql_free_result(res);
debug(D_NOTICE, "re-inserted some free pages into database, commiting transaction");
free_db_pages=cnt;
db_cache_max_page=cnt;
psync_sql_commit_transaction();
pthread_mutex_unlock(&flush_cache_mutex);
pthread_mutex_unlock(&clean_cache_mutex);
debug(D_NOTICE, "end");
}
int psync_pagecache_move_cache(const char *path){
psync_stat_t st;
psync_sql_res *res;
char *rdpath, *opath;
psync_file_t newrdcache, ordcache;
uint32_t i, cnt;
debug(D_NOTICE, "start");
rdpath=psync_strcat(path, PSYNC_DIRECTORY_SEPARATOR, PSYNC_DEFAULT_READ_CACHE_FILE, NULL);
if (!psync_stat(rdpath, &st)){
psync_free(rdpath);
return PRINT_RETURN_CONST(PERROR_CACHE_MOVE_NOT_EMPTY);
}
newrdcache=psync_file_open(rdpath, P_O_RDWR, P_O_CREAT|P_O_EXCL);
if (newrdcache==INVALID_HANDLE_VALUE){
psync_free(rdpath);
return PRINT_RETURN_CONST(PERROR_CACHE_MOVE_NO_WRITE_ACCESS);
}
opath=psync_strdup(psync_setting_get_string(_PS(fscachepath)));
pthread_mutex_lock(&clean_cache_mutex);
pthread_mutex_lock(&flush_cache_mutex);
psync_sql_start_transaction();
debug(D_NOTICE, "aquired locks");
if (psync_sql_cellint("SELECT COUNT(*) FROM fstask", 0)!=0){
if (IS_DEBUG){
psync_variant_row row;
debug(D_NOTICE, "the following tasks are preventing the cache move:");
res=psync_sql_query_nolock("SELECT id, type, status, folderid, text1 FROM fstask LIMIT 10");
while ((row=psync_sql_fetch_row(res)))
debug(D_NOTICE, "%u %u %u %u %s", (unsigned)psync_get_number(row[0]), (unsigned)psync_get_number(row[1]),
(unsigned)psync_get_number(row[2]), (unsigned)psync_get_number(row[3]), psync_get_string(row[4]));
}
psync_sql_rollback_transaction();
pthread_mutex_unlock(&flush_cache_mutex);
pthread_mutex_unlock(&clean_cache_mutex);
psync_file_close(newrdcache);
psync_file_delete(rdpath);
psync_free(opath);
psync_free(rdpath);
return PRINT_RETURN_CONST(PERROR_CACHE_MOVE_DRIVE_HAS_TASKS);
}
ordcache=readcache;
readcache=newrdcache;
psync_setting_set_string(_PS(fscachepath), path);
db_cache_in_pages=psync_setting_get_uint(_PS(fscachesize))/PSYNC_FS_PAGE_SIZE;
if (db_cache_in_pages<2*CACHE_PAGES)
cnt=db_cache_in_pages;
else
cnt=2*CACHE_PAGES;
psync_file_seek(readcache, cnt*PSYNC_FS_PAGE_SIZE, P_SEEK_SET);
assertw(psync_file_truncate(readcache)==0);
debug(D_NOTICE, "truncated cache file");
res=psync_sql_prep_statement("DELETE FROM pagecache");
psync_sql_run_free(res);
debug(D_NOTICE, "deleted entries from pagecache");
res=psync_sql_prep_statement("INSERT INTO pagecache (type) VALUES ("NTO_STR(PAGE_TYPE_FREE)")");
for (i=0; i<cnt; i++)
psync_sql_run(res);
psync_sql_free_result(res);
debug(D_NOTICE, "re-inserted some free pages into database, commiting transaction");
free_db_pages=cnt;
db_cache_max_page=cnt;
psync_sql_commit_transaction();
pthread_mutex_unlock(&flush_cache_mutex);
pthread_mutex_unlock(&clean_cache_mutex);
debug(D_NOTICE, "released locks");
psync_file_close(ordcache);
psync_list_dir(opath, clean_cache_del, (void *)1);
psync_free(opath);
psync_free(rdpath);
debug(D_NOTICE, "end");
return 0;
}