pdxx
/
fastdfs
mirror of https://github.com/happyfish100/fastdfs.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

storage nio use libserverframe

use_libserverframe
YuQing 2022-09-13 16:30:43 +08:00
parent bf3bfa68f6
commit d7c0594565
45 changed files with 509 additions and 1654 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
client/client_func.c
View File

@ -293,18 +293,18 @@ static int fdfs_client_do_init_ex(TrackerServerGroup *pTrackerGroup, \
}
}
g_fdfs_connect_timeout = iniGetIntValue(NULL, "connect_timeout", \
SF_G_CONNECT_TIMEOUT = iniGetIntValue(NULL, "connect_timeout", \
iniContext, DEFAULT_CONNECT_TIMEOUT);
if (g_fdfs_connect_timeout <= 0)
if (SF_G_CONNECT_TIMEOUT <= 0)
{
g_fdfs_connect_timeout = DEFAULT_CONNECT_TIMEOUT;
SF_G_CONNECT_TIMEOUT = DEFAULT_CONNECT_TIMEOUT;
}
g_fdfs_network_timeout = iniGetIntValue(NULL, "network_timeout", \
SF_G_NETWORK_TIMEOUT = iniGetIntValue(NULL, "network_timeout", \
iniContext, DEFAULT_NETWORK_TIMEOUT);
if (g_fdfs_network_timeout <= 0)
if (SF_G_NETWORK_TIMEOUT <= 0)
{
g_fdfs_network_timeout = DEFAULT_NETWORK_TIMEOUT;
SF_G_NETWORK_TIMEOUT = DEFAULT_NETWORK_TIMEOUT;
}
if ((result=fdfs_load_tracker_group_ex(pTrackerGroup, \
@ -376,8 +376,8 @@ static int fdfs_client_do_init_ex(TrackerServerGroup *pTrackerGroup, \
"use_connection_pool=%d, " \
"g_connection_pool_max_idle_time=%ds, " \
"use_storage_id=%d, storage server id count: %d\n", \
SF_G_BASE_PATH_STR, g_fdfs_connect_timeout, \
g_fdfs_network_timeout, pTrackerGroup->server_count, \
SF_G_BASE_PATH_STR, SF_G_CONNECT_TIMEOUT, \
SF_G_NETWORK_TIMEOUT, pTrackerGroup->server_count, \
g_anti_steal_token, g_anti_steal_secret_key.length, \
g_use_connection_pool, g_connection_pool_max_idle_time, \
use_storage_id, g_storage_ids_by_id.count);

1
client/client_global.c
View File

@ -15,4 +15,3 @@ TrackerServerGroup g_tracker_group = {0, 0, -1, NULL};
bool g_anti_steal_token = false;
BufferInfo g_anti_steal_secret_key = {0};

2
client/fdfs_appender_test.c
View File

@ -48,7 +48,7 @@ int uploadFileCallback(void *arg, const int64_t file_size, int sock)
filename = (char *)arg;
return tcpsendfile(sock, filename, file_size, \
g_fdfs_network_timeout, &total_send_bytes);
SF_G_NETWORK_TIMEOUT, &total_send_bytes);
}
int main(int argc, char *argv[])

2
client/fdfs_appender_test1.c
View File

@ -48,7 +48,7 @@ int uploadFileCallback(void *arg, const int64_t file_size, int sock)
filename = (char *)arg;
return tcpsendfile(sock, filename, file_size, \
g_fdfs_network_timeout, &total_send_bytes);
SF_G_NETWORK_TIMEOUT, &total_send_bytes);
}
int main(int argc, char *argv[])

2
client/fdfs_test.c
View File

@ -48,7 +48,7 @@ int uploadFileCallback(void *arg, const int64_t file_size, int sock)
filename = (char *)arg;
return tcpsendfile(sock, filename, file_size, \
g_fdfs_network_timeout, &total_send_bytes);
SF_G_NETWORK_TIMEOUT, &total_send_bytes);
}
int main(int argc, char *argv[])

2
client/fdfs_test1.c
View File

@ -47,7 +47,7 @@ int uploadFileCallback(void *arg, const int64_t file_size, int sock)
filename = (char *)arg;
return tcpsendfile(sock, filename, file_size, \
g_fdfs_network_timeout, &total_send_bytes);
SF_G_NETWORK_TIMEOUT, &total_send_bytes);
}
int main(int argc, char *argv[])

40
client/storage_client.c
View File

@ -252,7 +252,7 @@ int storage_get_metadata(ConnectionInfo *pTrackerServer, \
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN + \
filename_len, g_fdfs_network_timeout)) != 0)
filename_len, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -352,7 +352,7 @@ int storage_query_file_info_ex(ConnectionInfo *pTrackerServer, \
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN + \
filename_len, g_fdfs_network_timeout)) != 0)
filename_len, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -482,7 +482,7 @@ int storage_delete_file(ConnectionInfo *pTrackerServer, \
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN + \
filename_len, g_fdfs_network_timeout)) != 0)
filename_len, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -578,7 +578,7 @@ int storage_do_download_file_ex(ConnectionInfo *pTrackerServer, \
pHeader->cmd = STORAGE_PROTO_CMD_DOWNLOAD_FILE;
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
out_bytes, g_fdfs_network_timeout)) != 0)
out_bytes, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -601,7 +601,7 @@ int storage_do_download_file_ex(ConnectionInfo *pTrackerServer, \
if ((result=tcprecvfile(pStorageServer->sock, \
*file_buff, in_bytes, 0, \
g_fdfs_network_timeout, \
SF_G_NETWORK_TIMEOUT, \
&total_recv_bytes)) != 0)
{
break;
@ -649,7 +649,7 @@ int storage_do_download_file_ex(ConnectionInfo *pTrackerServer, \
}
if ((result=tcprecvdata_nb(pStorageServer->sock, buff, \
recv_bytes, g_fdfs_network_timeout)) != 0)
recv_bytes, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"recv data from storage server " \
@ -938,7 +938,7 @@ int storage_do_upload_file(ConnectionInfo *pTrackerServer, \
pHeader->status = 0;
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
p - out_buff, g_fdfs_network_timeout)) != 0)
p - out_buff, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -951,7 +951,7 @@ int storage_do_upload_file(ConnectionInfo *pTrackerServer, \
if (upload_type == FDFS_UPLOAD_BY_FILE)
{
if ((result=tcpsendfile(pStorageServer->sock, file_buff, \
file_size, g_fdfs_network_timeout, \
file_size, SF_G_NETWORK_TIMEOUT, \
&total_send_bytes)) != 0)
{
break;
@ -961,7 +961,7 @@ int storage_do_upload_file(ConnectionInfo *pTrackerServer, \
{
if ((result=tcpsenddata_nb(pStorageServer->sock, \
(char *)file_buff, file_size, \
g_fdfs_network_timeout)) != 0)
SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -1219,7 +1219,7 @@ int storage_set_metadata(ConnectionInfo *pTrackerServer, \
pHeader->cmd = STORAGE_PROTO_CMD_SET_METADATA;
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
p - out_buff, g_fdfs_network_timeout)) != 0)
p - out_buff, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -1231,7 +1231,7 @@ int storage_set_metadata(ConnectionInfo *pTrackerServer, \
}
if (meta_bytes > 0 && (result=tcpsenddata_nb(pStorageServer->sock, \
meta_buff, meta_bytes, g_fdfs_network_timeout)) != 0)
meta_buff, meta_bytes, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -1433,7 +1433,7 @@ int storage_client_create_link(ConnectionInfo *pTrackerServer, \
long2buff(p - out_buff - sizeof(TrackerHeader), pHeader->pkg_len);
pHeader->cmd = STORAGE_PROTO_CMD_CREATE_LINK;
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
p - out_buff, g_fdfs_network_timeout)) != 0)
p - out_buff, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -1729,7 +1729,7 @@ int storage_do_append_file(ConnectionInfo *pTrackerServer, \
pHeader->status = 0;
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
p - out_buff, g_fdfs_network_timeout)) != 0)
p - out_buff, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -1742,7 +1742,7 @@ int storage_do_append_file(ConnectionInfo *pTrackerServer, \
if (upload_type == FDFS_UPLOAD_BY_FILE)
{
if ((result=tcpsendfile(pStorageServer->sock, file_buff, \
file_size, g_fdfs_network_timeout, \
file_size, SF_G_NETWORK_TIMEOUT, \
&total_send_bytes)) != 0)
{
break;
@ -1752,7 +1752,7 @@ int storage_do_append_file(ConnectionInfo *pTrackerServer, \
{
if ((result=tcpsenddata_nb(pStorageServer->sock, \
(char *)file_buff, file_size, \
g_fdfs_network_timeout)) != 0)
SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -1860,7 +1860,7 @@ int storage_do_modify_file(ConnectionInfo *pTrackerServer, \
pHeader->status = 0;
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
p - out_buff, g_fdfs_network_timeout)) != 0)
p - out_buff, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -1873,7 +1873,7 @@ int storage_do_modify_file(ConnectionInfo *pTrackerServer, \
if (upload_type == FDFS_UPLOAD_BY_FILE)
{
if ((result=tcpsendfile(pStorageServer->sock, file_buff, \
file_size, g_fdfs_network_timeout, \
file_size, SF_G_NETWORK_TIMEOUT, \
&total_send_bytes)) != 0)
{
break;
@ -1883,7 +1883,7 @@ int storage_do_modify_file(ConnectionInfo *pTrackerServer, \
{
if ((result=tcpsenddata_nb(pStorageServer->sock, \
(char *)file_buff, file_size, \
g_fdfs_network_timeout)) != 0)
SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -2324,7 +2324,7 @@ int storage_truncate_file(ConnectionInfo *pTrackerServer, \
pHeader->status = 0;
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
p - out_buff, g_fdfs_network_timeout)) != 0)
p - out_buff, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -2400,7 +2400,7 @@ int storage_regenerate_appender_filename(ConnectionInfo *pTrackerServer,
pHeader->status = 0;
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff,
p - out_buff, g_fdfs_network_timeout)) != 0)
p - out_buff, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, "
"send data to storage server %s:%d fail, "

2
client/test/fdfs_test.c
View File

@ -48,7 +48,7 @@ int uploadFileCallback(void *arg, const int64_t file_size, int sock)
filename = (char *)arg;
return tcpsendfile(sock, filename, file_size, \
g_fdfs_network_timeout, &total_send_bytes);
SF_G_NETWORK_TIMEOUT, &total_send_bytes);
}
int main(int argc, char *argv[])

2
client/test/fdfs_test1.c
View File

@ -47,7 +47,7 @@ int uploadFileCallback(void *arg, const int64_t file_size, int sock)
filename = (char *)arg;
return tcpsendfile(sock, filename, file_size, \
g_fdfs_network_timeout, &total_send_bytes);
SF_G_NETWORK_TIMEOUT, &total_send_bytes);
}
int main(int argc, char *argv[])

26
client/tracker_client.c
View File

@ -295,7 +295,7 @@ int tracker_list_servers(ConnectionInfo *pTrackerServer, \
pHeader->cmd = TRACKER_PROTO_CMD_SERVER_LIST_STORAGE;
if ((result=tcpsenddata_nb(conn->sock, out_buff, \
sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN + id_len, \
g_fdfs_network_timeout)) != 0)
SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to tracker server %s:%d fail, " \
@ -498,7 +498,7 @@ int tracker_list_one_group(ConnectionInfo *pTrackerServer, \
pHeader->cmd = TRACKER_PROTO_CMD_SERVER_LIST_ONE_GROUP;
long2buff(FDFS_GROUP_NAME_MAX_LEN, pHeader->pkg_len);
if ((result=tcpsenddata_nb(conn->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to tracker server %s:%d fail, " \
@ -578,7 +578,7 @@ int tracker_list_groups(ConnectionInfo *pTrackerServer, \
header.cmd = TRACKER_PROTO_CMD_SERVER_LIST_ALL_GROUPS;
header.status = 0;
if ((result=tcpsenddata_nb(conn->sock, &header, \
sizeof(header), g_fdfs_network_timeout)) != 0)
sizeof(header), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to tracker server %s:%d fail, " \
@ -695,7 +695,7 @@ int tracker_do_query_storage(ConnectionInfo *pTrackerServer, \
pHeader->cmd = cmd;
if ((result=tcpsenddata_nb(conn->sock, out_buff, \
sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN +
filename_len, g_fdfs_network_timeout)) != 0)
filename_len, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to tracker server %s:%d fail, " \
@ -779,7 +779,7 @@ int tracker_query_storage_list(ConnectionInfo *pTrackerServer, \
pHeader->cmd = TRACKER_PROTO_CMD_SERVICE_QUERY_FETCH_ALL;
if ((result=tcpsenddata_nb(conn->sock, out_buff, \
sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN +
filename_len, g_fdfs_network_timeout)) != 0)
filename_len, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to tracker server %s:%d fail, " \
@ -878,7 +878,7 @@ int tracker_query_storage_store_without_group(ConnectionInfo *pTrackerServer,
memset(&header, 0, sizeof(header));
header.cmd = TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ONE;
if ((result=tcpsenddata_nb(conn->sock, &header, \
sizeof(header), g_fdfs_network_timeout)) != 0)
sizeof(header), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to tracker server %s:%d fail, " \
@ -961,7 +961,7 @@ int tracker_query_storage_store_with_group(ConnectionInfo *pTrackerServer, \
pHeader->cmd = TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITH_GROUP_ONE;
if ((result=tcpsenddata_nb(conn->sock, out_buff, \
sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN, \
g_fdfs_network_timeout)) != 0)
SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to tracker server %s:%d fail, " \
@ -1056,7 +1056,7 @@ int tracker_query_storage_store_list_with_group( \
long2buff(out_len, pHeader->pkg_len);
if ((result=tcpsenddata_nb(conn->sock, out_buff, \
sizeof(TrackerHeader) + out_len, g_fdfs_network_timeout)) != 0)
sizeof(TrackerHeader) + out_len, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to tracker server %s:%d fail, " \
@ -1226,7 +1226,7 @@ int tracker_delete_storage(TrackerServerGroup *pTrackerGroup, \
if ((result=tcpsenddata_nb(conn->sock, out_buff, \
sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN +
storage_id_len, g_fdfs_network_timeout)) != 0)
storage_id_len, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, "
"send data to tracker server %s:%d fail, "
@ -1306,7 +1306,7 @@ int tracker_delete_group(TrackerServerGroup *pTrackerGroup, \
if ((result=tcpsenddata_nb(conn->sock, out_buff,
sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN,
g_fdfs_network_timeout)) != 0)
SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, "
"send data to tracker server %s:%d fail, "
@ -1382,7 +1382,7 @@ int tracker_set_trunk_server(TrackerServerGroup *pTrackerGroup, \
if ((result=tcpsenddata_nb(conn->sock, out_buff, \
sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN +
storage_id_len, g_fdfs_network_timeout)) != 0)
storage_id_len, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, "
"send data to tracker server %s:%d fail, "
@ -1468,7 +1468,7 @@ int tracker_get_storage_status(ConnectionInfo *pTrackerServer,
pHeader->cmd = TRACKER_PROTO_CMD_STORAGE_GET_STATUS;
long2buff(FDFS_GROUP_NAME_MAX_LEN + ip_len, pHeader->pkg_len);
if ((result=tcpsenddata_nb(conn->sock, out_buff, \
p - out_buff, g_fdfs_network_timeout)) != 0)
p - out_buff, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to tracker server %s:%d fail, " \
@ -1556,7 +1556,7 @@ int tracker_get_storage_id(ConnectionInfo *pTrackerServer, \
pHeader->cmd = TRACKER_PROTO_CMD_STORAGE_GET_SERVER_ID;
long2buff(FDFS_GROUP_NAME_MAX_LEN + ip_len, pHeader->pkg_len);
if ((result=tcpsenddata_nb(conn->sock, out_buff, \
p - out_buff, g_fdfs_network_timeout)) != 0)
p - out_buff, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to tracker server %s:%d fail, " \

2
common/fdfs_global.c
View File

@ -20,8 +20,6 @@
#include "fastcommon/logger.h"
#include "fdfs_global.h"
int g_fdfs_connect_timeout = DEFAULT_CONNECT_TIMEOUT;
int g_fdfs_network_timeout = DEFAULT_NETWORK_TIMEOUT;
Version g_fdfs_version = {6, 9, 0};
bool g_use_connection_pool = false;
ConnectionPool g_connection_pool;

2
common/fdfs_global.h
View File

@ -22,8 +22,6 @@
extern "C" {
#endif
extern int g_fdfs_connect_timeout;
extern int g_fdfs_network_timeout;
extern Version g_fdfs_version;
extern bool g_use_connection_pool;
extern ConnectionPool g_connection_pool;

6
fastdfs.spec
View File

@ -19,7 +19,7 @@ BuildRoot: %{_tmppath}/%{name}-%{version}-%{release}-root-%(%{__id_u} -n)
Requires: %__cp %__mv %__chmod %__grep %__mkdir %__install %__id
BuildRequires: libfastcommon-devel >= 1.0.60
BuildRequires: libserverframe-devel >= 1.1.18
BuildRequires: libserverframe-devel >= 1.1.19
%description
This package provides tracker & storage of fastdfs
@ -27,7 +27,7 @@ commit version: %{CommitVersion}
%package -n %{FDFSServer}
Requires: libfastcommon >= 1.0.60
Requires: libserverframe >= 1.1.18
Requires: libserverframe >= 1.1.19
Requires: %{FDFSConfig}
Summary: fastdfs tracker & storage
@ -37,7 +37,7 @@ Summary: fastdfs tools
%package -n %{FDFSClient}
Requires: libfastcommon >= 1.0.60
Requires: libserverframe >= 1.1.18
Requires: libserverframe >= 1.1.19
Requires: %{FDFSConfig}
Summary: The client dynamic library of fastdfs

2
storage/Makefile.in
View File

@ -10,7 +10,7 @@ SHARED_OBJS = ../common/fdfs_global.o ../tracker/fdfs_shared_func.o \
../tracker/fdfs_server_id_func.o ../tracker/tracker_proto.o \
tracker_client_thread.o storage_global.o storage_func.o \
storage_sync_func.o storage_service.o storage_sync.o \
storage_nio.o storage_dio.o storage_ip_changed_dealer.o \
storage_dio.o storage_ip_changed_dealer.o \
storage_param_getter.o storage_disk_recovery.o \
trunk_mgr/trunk_mem.o trunk_mgr/trunk_shared.o \
trunk_mgr/trunk_sync.o trunk_mgr/trunk_client.o \

92
storage/fdfs_storaged.c
View File

@ -64,7 +64,7 @@ static void sigHupHandler(int sig);
static void sigUsrHandler(int sig);
static void sigAlarmHandler(int sig);
static int setupSchedules(pthread_t *schedule_tid);
static int setup_schedule_tasks();
static int setupSignalHandlers();
#if defined(DEBUG_FLAG)
@ -83,7 +83,6 @@ int main(int argc, char *argv[])
const char *conf_filename;
char *action;
int result;
int sock;
int wait_count;
pthread_t schedule_tid;
char pidFilename[MAX_PATH_SIZE];
@ -103,7 +102,6 @@ int main(int argc, char *argv[])
}
g_current_time = time(NULL);
g_up_time = g_current_time;
log_init2();
if ((result=trunk_shared_init()) != 0)
{
@ -155,12 +153,6 @@ int main(int argc, char *argv[])
}
umask(0);
if ((result=write_to_pid_file(pidFilename)) != 0)
{
log_destroy();
return result;
}
if ((result=setupSignalHandlers()) != 0)
{
logCrit("exit abnormally!\n");
@ -168,29 +160,21 @@ int main(int argc, char *argv[])
return result;
}
memset(SF_G_INNER_BIND_ADDR, 0, sizeof(SF_G_INNER_BIND_ADDR));
if ((result=storage_func_init(conf_filename, \
SF_G_INNER_BIND_ADDR, sizeof(SF_G_INNER_BIND_ADDR))) != 0)
if ((result=storage_func_init(conf_filename)) != 0)
{
logCrit("exit abnormally!\n");
delete_pid_file(pidFilename);
log_destroy();
return result;
}
sock = socketServer(SF_G_INNER_BIND_ADDR, SF_G_INNER_PORT, &result);
if (sock < 0)
if ((result=sf_socket_server()) != 0)
{
logCrit("exit abnormally!\n");
delete_pid_file(pidFilename);
log_destroy();
return result;
}
if ((result=tcpsetserveropt(sock, g_fdfs_network_timeout)) != 0)
if ((result=write_to_pid_file(pidFilename)) != 0)
{
logCrit("exit abnormally!\n");
delete_pid_file(pidFilename);
log_destroy();
return result;
}
@ -252,14 +236,22 @@ int main(int argc, char *argv[])
return result;
}
if ((result=setupSchedules(&schedule_tid)) != 0)
if ((result=sf_startup_schedule(&schedule_tid)) != 0)
{
log_destroy();
return result;
}
if ((result=setup_schedule_tasks()) != 0)
{
logCrit("exit abnormally!\n");
log_destroy();
return result;
}
if ((result=set_run_by(g_run_by_group, g_run_by_user)) != 0)
if ((result=set_run_by(g_sf_global_vars.run_by_group,
g_sf_global_vars.run_by_user)) != 0)
{
logCrit("exit abnormally!\n");
log_destroy();
@ -277,7 +269,7 @@ int main(int argc, char *argv[])
bTerminateFlag = false;
accept_stage = ACCEPT_STAGE_DOING;
storage_accept_loop(sock);
sf_accept_loop();
accept_stage = ACCEPT_STAGE_DONE;
fdfs_binlog_sync_func(NULL); //binlog fsync
@ -287,17 +279,14 @@ int main(int argc, char *argv[])
pthread_kill(schedule_tid, SIGINT);
}
storage_terminate_threads();
storage_dio_terminate();
kill_tracker_report_threads();
kill_storage_sync_threads();
wait_count = 0;
while (g_storage_thread_count != 0 || \
g_dio_thread_count != 0 || \
g_tracker_reporter_count > 0 || \
g_schedule_flag)
while (SF_G_ALIVE_THREAD_COUNT != 0 || g_dio_thread_count != 0 ||
g_tracker_reporter_count > 0 || g_schedule_flag)
{
/*
#if defined(DEBUG_FLAG) && defined(OS_LINUX)
@ -374,7 +363,7 @@ static void sigAlarmHandler(int sig)
server.port = SF_G_INNER_PORT;
server.sock = -1;
if (conn_pool_connect_server(&server, g_fdfs_connect_timeout) != 0)
if (conn_pool_connect_server(&server, SF_G_CONNECT_TIMEOUT) != 0)
{
return;
}
@ -388,7 +377,7 @@ static void sigAlarmHandler(int sig)
static void sigHupHandler(int sig)
{
if (g_rotate_error_log)
if (g_sf_global_vars.error_log.rotate_everyday)
{
g_log_context.rotate_immediately = true;
}
@ -429,23 +418,17 @@ static void sigDumpHandler(int sig)
}
#endif
static int setupSchedules(pthread_t *schedule_tid)
static int setup_schedule_tasks()
{
#define SCHEDULE_ENTRIES_MAX_COUNT 10
#define SCHEDULE_ENTRIES_MAX_COUNT 8
ScheduleEntry scheduleEntries[SCHEDULE_ENTRIES_MAX_COUNT];
ScheduleArray scheduleArray;
int result;
scheduleArray.entries = scheduleEntries;
scheduleArray.count = 0;
memset(scheduleEntries, 0, sizeof(scheduleEntries));
INIT_SCHEDULE_ENTRY(scheduleEntries[scheduleArray.count],
scheduleArray.count + 1, TIME_NONE, TIME_NONE, TIME_NONE,
g_sync_log_buff_interval, log_sync_func, &g_log_context);
scheduleArray.count++;
INIT_SCHEDULE_ENTRY(scheduleEntries[scheduleArray.count],
scheduleArray.count + 1, TIME_NONE, TIME_NONE, TIME_NONE,
g_sync_binlog_buff_interval, fdfs_binlog_sync_func, NULL);
@ -468,7 +451,8 @@ static int setupSchedules(pthread_t *schedule_tid)
{
INIT_SCHEDULE_ENTRY(scheduleEntries[scheduleArray.count],
scheduleArray.count + 1, TIME_NONE, TIME_NONE, TIME_NONE,
g_sync_log_buff_interval, log_sync_func, &g_access_log_context);
g_sf_global_vars.error_log.sync_log_buff_interval,
log_sync_func, &g_access_log_context);
scheduleArray.count++;
if (g_rotate_access_log)
@ -478,10 +462,10 @@ static int setupSchedules(pthread_t *schedule_tid)
24 * 3600, log_notify_rotate, &g_access_log_context);
scheduleArray.count++;
if (g_log_file_keep_days > 0)
if (g_sf_global_vars.error_log.keep_days > 0)
{
log_set_keep_days(&g_access_log_context,
g_log_file_keep_days);
g_sf_global_vars.error_log.keep_days);
INIT_SCHEDULE_ENTRY(scheduleEntries[scheduleArray.count],
scheduleArray.count + 1, 1, 0, 0, 24 * 3600,
@ -491,24 +475,6 @@ static int setupSchedules(pthread_t *schedule_tid)
}
}
if (g_rotate_error_log)
{
INIT_SCHEDULE_ENTRY_EX(scheduleEntries[scheduleArray.count],
scheduleArray.count + 1, g_error_log_rotate_time,
24 * 3600, log_notify_rotate, &g_log_context);
scheduleArray.count++;
if (g_log_file_keep_days > 0)
{
log_set_keep_days(&g_log_context, g_log_file_keep_days);
INIT_SCHEDULE_ENTRY(scheduleEntries[scheduleArray.count],
scheduleArray.count + 1, 1, 0, 0, 24 * 3600,
log_delete_old_files, &g_log_context);
scheduleArray.count++;
}
}
if (g_compress_binlog)
{
INIT_SCHEDULE_ENTRY_EX(scheduleEntries[scheduleArray.count],
@ -517,13 +483,7 @@ static int setupSchedules(pthread_t *schedule_tid)
scheduleArray.count++;
}
if ((result=sched_start(&scheduleArray, schedule_tid,
SF_G_THREAD_STACK_SIZE, (bool * volatile)&SF_G_CONTINUE_FLAG)) != 0)
{
return result;
}
return 0;
return sched_add_entries(&scheduleArray);
}
static int setupSignalHandlers()

12
storage/storage_dio.c
View File

@ -25,10 +25,10 @@
#include "fastcommon/logger.h"
#include "fastcommon/sockopt.h"
#include "fastcommon/ioevent_loop.h"
#include "storage_dio.h"
#include "storage_nio.h"
#include "storage_global.h"
#include "storage_service.h"
#include "trunk_mem.h"
#include "storage_dio.h"
static pthread_mutex_t g_dio_thread_lock;
static struct storage_dio_context *g_dio_contexts = NULL;
@ -153,8 +153,6 @@ int storage_dio_queue_push(struct fast_task_info *pTask)
pClientInfo = (StorageClientInfo *)pTask->arg;
pFileContext = &(pClientInfo->file_context);
pContext = g_dio_contexts + pFileContext->dio_thread_index;
pClientInfo->stage |= FDFS_STORAGE_STAGE_DIO_THREAD;
if ((result=blocked_queue_push(&(pContext->queue), pTask)) != 0)
{
ioevent_add_to_deleted_list(pTask);
@ -246,7 +244,7 @@ int dio_discard_file(struct fast_task_info *pTask)
else
{
pFileContext->buff_offset = 0;
pFileContext->continue_callback(pTask);
pFileContext->continue_callback(pTask, SF_NIO_STAGE_RECV);
}
return 0;
@ -370,7 +368,7 @@ int dio_read_file(struct fast_task_info *pTask)
if (pFileContext->offset < pFileContext->end)
{
pFileContext->continue_callback(pTask);
pFileContext->continue_callback(pTask, SF_NIO_STAGE_SEND);
}
else
{
@ -487,7 +485,7 @@ int dio_write_file(struct fast_task_info *pTask)
if (pFileContext->offset < pFileContext->end)
{
pFileContext->buff_offset = 0;
pFileContext->continue_callback(pTask);
pFileContext->continue_callback(pTask, SF_NIO_STAGE_RECV);
}
else
{

6
storage/storage_disk_recovery.c
View File

@ -140,7 +140,7 @@ static int storage_do_fetch_binlog(ConnectionInfo *pSrcStorage, \
store_path_index;
if((result=tcpsenddata_nb(pSrcStorage->sock, out_buff,
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, "
"storage server %s:%d, send data fail, "
@ -150,9 +150,9 @@ static int storage_do_fetch_binlog(ConnectionInfo *pSrcStorage, \
return result;
}
if (g_fdfs_network_timeout >= 600)
if (SF_G_NETWORK_TIMEOUT >= 600)
{
network_timeout = g_fdfs_network_timeout;
network_timeout = SF_G_NETWORK_TIMEOUT;
}
else
{

22
storage/storage_dump.c
View File

@ -42,8 +42,8 @@ static int fdfs_dump_global_vars(char *buff, const int buffSize)
last_storage_ip_str, sizeof(last_storage_ip_str));
total_len = snprintf(buff, buffSize,
"g_fdfs_connect_timeout=%ds\n"
"g_fdfs_network_timeout=%ds\n"
"SF_G_CONNECT_TIMEOUT=%ds\n"
"SF_G_NETWORK_TIMEOUT=%ds\n"
"SF_G_BASE_PATH_STR=%s\n"
"g_fdfs_version=%d.%02d\n"
"SF_G_CONTINUE_FLAG=%d\n"
@ -52,7 +52,6 @@ static int fdfs_dump_global_vars(char *buff, const int buffSize)
"g_max_connections=%d\n"
"g_storage_thread_count=%d\n"
"g_group_name=%s\n"
"g_sync_log_buff_interval=%d\n"
"g_subdir_count_per_path=%d\n"
"g_http_port=%d\n"
"g_last_server_port=%d\n"
@ -136,24 +135,23 @@ static int fdfs_dump_global_vars(char *buff, const int buffSize)
#if defined(DEBUG_FLAG) && defined(OS_LINUX)
"g_exe_name=%s\n"
#endif
, g_fdfs_connect_timeout
, g_fdfs_network_timeout
, SF_G_CONNECT_TIMEOUT
, SF_G_NETWORK_TIMEOUT
, SF_G_BASE_PATH_STR
, g_fdfs_version.major, g_fdfs_version.minor
, SF_G_CONTINUE_FLAG
, g_schedule_flag
, SF_G_INNER_PORT
, g_max_connections
, g_storage_thread_count
, g_sf_global_vars.max_connections
, SF_G_ALIVE_THREAD_COUNT
, g_group_name
, g_sync_log_buff_interval
, g_subdir_count_per_path
, g_http_port
, g_last_server_port
, g_last_http_port
, g_allow_ip_count
, g_run_by_group
, g_run_by_user
, g_sf_global_vars.run_by_group
, g_sf_global_vars.run_by_user
, g_http_domain
, g_file_distribute_path_mode
, g_file_distribute_rotate_count
@ -174,7 +172,7 @@ static int fdfs_dump_global_vars(char *buff, const int buffSize)
, g_sync_start_time.hour, g_sync_start_time.minute
, g_sync_end_time.hour, g_sync_end_time.minute
, g_sync_part_time
, g_sync_log_buff_interval
, g_sf_global_vars.error_log.sync_log_buff_interval
, g_sync_binlog_buff_interval
, g_write_mark_file_freq
, g_sync_stat_file_interval
@ -196,7 +194,7 @@ static int fdfs_dump_global_vars(char *buff, const int buffSize)
, g_thread_kill_done
, SF_G_THREAD_STACK_SIZE
, g_upload_priority
, formatDatetime(g_up_time, "%Y-%m-%d %H:%M:%S",
, formatDatetime(g_sf_global_vars.up_time, "%Y-%m-%d %H:%M:%S",
szUptime, sizeof(szUptime))
, g_if_alias_prefix
, g_binlog_fd

403
storage/storage_func.c
View File

@ -31,6 +31,7 @@
#include "fastcommon/sched_thread.h"
#include "fastcommon/ini_file_reader.h"
#include "fastcommon/connection_pool.h"
#include "sf/sf_service.h"
#include "tracker_types.h"
#include "tracker_proto.h"
#include "fdfs_shared_func.h"
@ -151,7 +152,7 @@ static int storage_do_get_group_name(ConnectionInfo *pTrackerServer)
int2buff(SF_G_INNER_PORT, out_buff + sizeof(TrackerHeader));
pHeader->cmd = TRACKER_PROTO_CMD_STORAGE_GET_GROUP_NAME;
if ((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \
@ -1400,27 +1401,22 @@ static int init_my_result_per_tracker()
return 0;
}
int storage_func_init(const char *filename, \
char *bind_addr, const int addr_size)
int storage_func_init(const char *filename)
{
char *pBindAddr;
const int task_buffer_extra_size = 0;
const bool need_set_run_by = false;
char *pGroupName;
char *pRunByGroup;
char *pRunByUser;
char *pFsyncAfterWrittenBytes;
char *pThreadStackSize;
char *pBuffSize;
char *pIfAliasPrefix;
char *pHttpDomain;
char *pRotateAccessLogSize;
char *pRotateErrorLogSize;
IniContext iniContext;
SFContextIniConfig config;
int result;
int64_t fsync_after_written_bytes;
int64_t thread_stack_size;
int64_t buff_size;
int64_t rotate_access_log_size;
int64_t rotate_error_log_size;
char sz_global_config[512];
char sz_service_config[128];
if ((result=iniLoadFromFile(filename, &iniContext)) != 0)
{
@ -1441,6 +1437,18 @@ int storage_func_init(const char *filename, \
break;
}
sf_set_current_time();
SF_SET_CONTEXT_INI_CONFIG_EX(config, filename, &iniContext,
NULL, FDFS_STORAGE_SERVER_DEF_PORT,
FDFS_STORAGE_SERVER_DEF_PORT, DEFAULT_WORK_THREADS,
"buff_size");
if ((result=sf_load_config_ex("storaged", &config,
task_buffer_extra_size, need_set_run_by)) != 0)
{
return result;
}
g_subdir_count_per_path=iniGetIntValue(NULL, \
"subdir_count_per_path", &iniContext, \
DEFAULT_DATA_DIR_COUNT_PER_PATH);
@ -1466,18 +1474,18 @@ int storage_func_init(const char *filename, \
break;
}
g_fdfs_connect_timeout = iniGetIntValue(NULL, "connect_timeout", \
SF_G_CONNECT_TIMEOUT = iniGetIntValue(NULL, "connect_timeout", \
&iniContext, DEFAULT_CONNECT_TIMEOUT);
if (g_fdfs_connect_timeout <= 0)
if (SF_G_CONNECT_TIMEOUT <= 0)
{
g_fdfs_connect_timeout = DEFAULT_CONNECT_TIMEOUT;
SF_G_CONNECT_TIMEOUT = DEFAULT_CONNECT_TIMEOUT;
}
g_fdfs_network_timeout = iniGetIntValue(NULL, "network_timeout", \
SF_G_NETWORK_TIMEOUT = iniGetIntValue(NULL, "network_timeout", \
&iniContext, DEFAULT_NETWORK_TIMEOUT);
if (g_fdfs_network_timeout <= 0)
if (SF_G_NETWORK_TIMEOUT <= 0)
{
g_fdfs_network_timeout = DEFAULT_NETWORK_TIMEOUT;
SF_G_NETWORK_TIMEOUT = DEFAULT_NETWORK_TIMEOUT;
}
SF_G_INNER_PORT = iniGetIntValue(NULL, "port", &iniContext, \
@ -1503,16 +1511,6 @@ int storage_func_init(const char *filename, \
g_stat_report_interval = STORAGE_REPORT_DEF_INTERVAL;
}
pBindAddr = iniGetStrValue(NULL, "bind_addr", &iniContext);
if (pBindAddr == NULL)
{
*bind_addr = '\0';
}
else
{
snprintf(bind_addr, addr_size, "%s", pBindAddr);
}
g_client_bind_addr = iniGetBoolValue(NULL, "client_bind", \
&iniContext, true);
@ -1604,60 +1602,13 @@ int storage_func_init(const char *filename, \
(g_sync_end_time.hour == 23 && \
g_sync_end_time.minute == 59));
g_max_connections = iniGetIntValue(NULL, "max_connections", \
&iniContext, DEFAULT_MAX_CONNECTONS);
if (g_max_connections <= 0)
{
g_max_connections = DEFAULT_MAX_CONNECTONS;
}
if ((result=set_rlimit(RLIMIT_NOFILE, g_max_connections)) != 0)
{
break;
}
g_accept_threads = iniGetIntValue(NULL, "accept_threads", \
&iniContext, 1);
if (g_accept_threads <= 0)
{
logError("file: "__FILE__", line: %d, " \
"item \"accept_threads\" is invalid, " \
"value: %d <= 0!", __LINE__, g_accept_threads);
result = EINVAL;
break;
}
g_work_threads = iniGetIntValue(NULL, "work_threads", \
&iniContext, DEFAULT_WORK_THREADS);
if (g_work_threads <= 0)
{
logError("file: "__FILE__", line: %d, " \
"item \"work_threads\" is invalid, " \
"value: %d <= 0!", __LINE__, g_work_threads);
result = EINVAL;
break;
}
pBuffSize = iniGetStrValue(NULL, \
"buff_size", &iniContext);
if (pBuffSize == NULL)
{
buff_size = STORAGE_DEFAULT_BUFF_SIZE;
}
else if ((result=parse_bytes(pBuffSize, 1, &buff_size)) != 0)
{
break;
}
g_buff_size = buff_size;
if (g_buff_size < 4 * 1024 || \
g_buff_size < sizeof(TrackerHeader) + \
if (g_sf_global_vars.min_buff_size < sizeof(TrackerHeader) +
TRUNK_BINLOG_BUFFER_SIZE)
{
logError("file: "__FILE__", line: %d, " \
"item \"buff_size\" is too small, " \
"value: %d < %d or < %d!", __LINE__, \
g_buff_size, 4 * 1024, \
(int)sizeof(TrackerHeader) + \
TRUNK_BINLOG_BUFFER_SIZE);
logError("file: "__FILE__", line: %d, "
"item \"buff_size\" is too small, value: %d < %d!",
__LINE__, g_sf_global_vars.min_buff_size,
(int)sizeof(TrackerHeader) + TRUNK_BINLOG_BUFFER_SIZE);
result = EINVAL;
break;
}
@ -1733,75 +1684,6 @@ int storage_func_init(const char *filename, \
break;
}
/*
g_disk_rw_direct = iniGetBoolValue(NULL, \
"disk_rw_direct", &iniContext, false);
*/
pRunByGroup = iniGetStrValue(NULL, "run_by_group", &iniContext);
pRunByUser = iniGetStrValue(NULL, "run_by_user", &iniContext);
if (pRunByGroup == NULL)
{
*g_run_by_group = '\0';
}
else
{
snprintf(g_run_by_group, sizeof(g_run_by_group), \
"%s", pRunByGroup);
}
if (*g_run_by_group == '\0')
{
g_run_by_gid = getegid();
}
else
{
struct group *pGroup;
pGroup = getgrnam(g_run_by_group);
if (pGroup == NULL)
{
result = errno != 0 ? errno : ENOENT;
logError("file: "__FILE__", line: %d, " \
"getgrnam fail, errno: %d, " \
"error info: %s", __LINE__, \
result, STRERROR(result));
return result;
}
g_run_by_gid = pGroup->gr_gid;
}
if (pRunByUser == NULL)
{
*g_run_by_user = '\0';
}
else
{
snprintf(g_run_by_user, sizeof(g_run_by_user), \
"%s", pRunByUser);
}
if (*g_run_by_user == '\0')
{
g_run_by_uid = geteuid();
}
else
{
struct passwd *pUser;
pUser = getpwnam(g_run_by_user);
if (pUser == NULL)
{
result = errno != 0 ? errno : ENOENT;
logError("file: "__FILE__", line: %d, " \
"getpwnam fail, errno: %d, " \
"error info: %s", __LINE__, \
result, STRERROR(result));
return result;
}
g_run_by_uid = pUser->pw_uid;
}
if ((result=load_allow_hosts(&iniContext, \
&g_allow_ip_addrs, &g_allow_ip_count)) != 0)
{
@ -1833,14 +1715,6 @@ int storage_func_init(const char *filename, \
}
g_fsync_after_written_bytes = fsync_after_written_bytes;
g_sync_log_buff_interval = iniGetIntValue(NULL, \
"sync_log_buff_interval", &iniContext, \
SYNC_LOG_BUFF_DEF_INTERVAL);
if (g_sync_log_buff_interval <= 0)
{
g_sync_log_buff_interval = SYNC_LOG_BUFF_DEF_INTERVAL;
}
g_sync_binlog_buff_interval = iniGetIntValue(NULL, \
"sync_binlog_buff_interval", &iniContext,\
SYNC_BINLOG_BUFF_DEF_INTERVAL);
@ -1866,19 +1740,6 @@ int storage_func_init(const char *filename, \
g_sync_stat_file_interval=DEFAULT_SYNC_STAT_FILE_INTERVAL;
}
pThreadStackSize = iniGetStrValue(NULL, \
"thread_stack_size", &iniContext);
if (pThreadStackSize == NULL)
{
thread_stack_size = 512 * 1024;
}
else if ((result=parse_bytes(pThreadStackSize, 1, \
&thread_stack_size)) != 0)
{
break;
}
SF_G_THREAD_STACK_SIZE = (int)thread_stack_size;
if (SF_G_THREAD_STACK_SIZE < FAST_WRITE_BUFF_SIZE + 64 * 1024)
{
logError("file: "__FILE__", line: %d, " \
@ -1927,8 +1788,8 @@ int storage_func_init(const char *filename, \
}
strcpy(g_fdht_base_path, SF_G_BASE_PATH_STR);
g_fdht_connect_timeout = g_fdfs_connect_timeout;
g_fdht_network_timeout = g_fdfs_network_timeout;
g_fdht_connect_timeout = SF_G_CONNECT_TIMEOUT;
g_fdht_network_timeout = SF_G_NETWORK_TIMEOUT;
pKeyNamespace = iniGetStrValue(NULL, \
"key_namespace", &iniContext);
@ -2013,23 +1874,10 @@ int storage_func_init(const char *filename, \
break;
}
g_rotate_error_log = iniGetBoolValue(NULL, "rotate_error_log",\
&iniContext, false);
g_compress_old_access_log = iniGetBoolValue(NULL, "compress_old_access_log",
&iniContext, false);
g_compress_old_error_log = iniGetBoolValue(NULL, "compress_old_error_log",
&iniContext, false);
g_compress_error_log_days_before = iniGetIntValue(NULL,
"compress_error_log_days_before", &iniContext, 1);
g_compress_old_access_log = iniGetBoolValue(NULL,
"compress_old_access_log", &iniContext, false);
g_compress_access_log_days_before = iniGetIntValue(NULL,
"compress_access_log_days_before", &iniContext, 1);
if (g_compress_old_error_log)
{
log_set_compress_log_flags(LOG_COMPRESS_FLAGS_ENABLED |
LOG_COMPRESS_FLAGS_NEW_THREAD);
log_set_compress_log_days_before(g_compress_error_log_days_before);
}
if (g_use_access_log && g_compress_old_access_log)
{
log_set_compress_log_flags_ex(&g_access_log_context,
@ -2039,13 +1887,6 @@ int storage_func_init(const char *filename, \
g_compress_access_log_days_before);
}
if ((result=get_time_item_from_conf(&iniContext, \
"error_log_rotate_time", &g_error_log_rotate_time, \
0, 0)) != 0)
{
break;
}
pRotateAccessLogSize = iniGetStrValue(NULL, \
"rotate_access_log_size", &iniContext);
if (pRotateAccessLogSize == NULL)
@ -2069,32 +1910,6 @@ int storage_func_init(const char *filename, \
}
fdfs_set_log_rotate_size(&g_access_log_context, rotate_access_log_size);
pRotateErrorLogSize = iniGetStrValue(NULL, \
"rotate_error_log_size", &iniContext);
if (pRotateErrorLogSize == NULL)
{
rotate_error_log_size = 0;
}
else if ((result=parse_bytes(pRotateErrorLogSize, 1, \
&rotate_error_log_size)) != 0)
{
break;
}
if (rotate_error_log_size > 0 && \
rotate_error_log_size < FDFS_ONE_MB)
{
logWarning("file: "__FILE__", line: %d, " \
"item \"rotate_error_log_size\": " \
"%"PRId64" is too small, " \
"change to 1 MB", __LINE__, \
rotate_error_log_size);
rotate_error_log_size = FDFS_ONE_MB;
}
fdfs_set_log_rotate_size(&g_log_context, rotate_error_log_size);
g_log_file_keep_days = iniGetIntValue(NULL, \
"log_file_keep_days", &iniContext, 0);
g_file_sync_skip_invalid_record = iniGetBoolValue(NULL, \
"file_sync_skip_invalid_record", &iniContext, false);
@ -2140,88 +1955,70 @@ int storage_func_init(const char *filename, \
}
#endif
logInfo("FastDFS v%d.%02d, base_path=%s, store_path_count=%d, " \
"subdir_count_per_path=%d, group_name=%s, " \
"run_by_group=%s, run_by_user=%s, " \
"connect_timeout=%ds, network_timeout=%ds, "\
"port=%d, bind_addr=%s, client_bind=%d, " \
"max_connections=%d, accept_threads=%d, " \
"work_threads=%d, " \
"disk_rw_separated=%d, disk_reader_threads=%d, " \
"disk_writer_threads=%d, disk_recovery_threads=%d, " \
"buff_size=%d KB, heart_beat_interval=%ds, " \
"stat_report_interval=%ds, tracker_server_count=%d, " \
"sync_wait_msec=%dms, sync_interval=%dms, " \
"sync_start_time=%02d:%02d, sync_end_time=%02d:%02d, "\
"write_mark_file_freq=%d, " \
"allow_ip_count=%d, " \
"file_distribute_path_mode=%d, " \
"file_distribute_rotate_count=%d, " \
"fsync_after_written_bytes=%d, " \
"sync_log_buff_interval=%ds, " \
"sync_binlog_buff_interval=%ds, " \
"sync_stat_file_interval=%ds, " \
"thread_stack_size=%d KB, upload_priority=%d, " \
"if_alias_prefix=%s, " \
"check_file_duplicate=%d, file_signature_method=%s, " \
"FDHT group count=%d, FDHT server count=%d, " \
"FDHT key_namespace=%s, FDHT keep_alive=%d, " \
"HTTP server port=%d, domain name=%s, " \
"use_access_log=%d, rotate_access_log=%d, " \
"access_log_rotate_time=%02d:%02d, " \
"compress_old_access_log=%d, " \
"compress_access_log_days_before=%d, " \
"rotate_error_log=%d, " \
"error_log_rotate_time=%02d:%02d, " \
"compress_old_error_log=%d, " \
"compress_error_log_days_before=%d, " \
"rotate_access_log_size=%"PRId64", " \
"rotate_error_log_size=%"PRId64", " \
"log_file_keep_days=%d, " \
"file_sync_skip_invalid_record=%d, " \
"use_connection_pool=%d, " \
"g_connection_pool_max_idle_time=%ds, " \
"compress_binlog=%d, " \
"compress_binlog_time=%02d:%02d, " \
"check_store_path_mark=%d", \
g_fdfs_version.major, g_fdfs_version.minor, \
SF_G_BASE_PATH_STR, g_fdfs_store_paths.count, \
g_subdir_count_per_path, \
g_group_name, g_run_by_group, g_run_by_user, \
g_fdfs_connect_timeout, g_fdfs_network_timeout, \
SF_G_INNER_PORT, bind_addr, \
g_client_bind_addr, g_max_connections, \
g_accept_threads, g_work_threads, g_disk_rw_separated, \
g_disk_reader_threads, g_disk_writer_threads, \
g_disk_recovery_threads, g_buff_size / 1024, \
g_heart_beat_interval, g_stat_report_interval, \
g_tracker_group.server_count, g_sync_wait_usec / 1000, \
g_sync_interval / 1000, \
g_sync_start_time.hour, g_sync_start_time.minute, \
g_sync_end_time.hour, g_sync_end_time.minute, \
g_write_mark_file_freq, \
g_allow_ip_count, g_file_distribute_path_mode, \
g_file_distribute_rotate_count, \
g_fsync_after_written_bytes, g_sync_log_buff_interval, \
g_sync_binlog_buff_interval, g_sync_stat_file_interval, \
SF_G_THREAD_STACK_SIZE/1024, g_upload_priority, \
g_if_alias_prefix, g_check_file_duplicate, \
g_file_signature_method == STORAGE_FILE_SIGNATURE_METHOD_HASH \
sf_global_config_to_string_ex("buff_size", sz_global_config,
sizeof(sz_global_config));
sf_context_config_to_string(&g_sf_context,
sz_service_config, sizeof(sz_service_config));
logInfo("FastDFS v%d.%02d, %s, %s, store_path_count=%d, "
"subdir_count_per_path=%d, group_name=%s, client_bind=%d, "
"disk_rw_separated=%d, disk_reader_threads=%d, "
"disk_writer_threads=%d, disk_recovery_threads=%d, "
"heart_beat_interval=%ds, stat_report_interval=%ds, "
"tracker_server_count=%d, sync_wait_msec=%dms, "
"sync_interval=%dms, sync_start_time=%02d:%02d, "
"sync_end_time=%02d:%02d, write_mark_file_freq=%d, "
"allow_ip_count=%d, "
"file_distribute_path_mode=%d, "
"file_distribute_rotate_count=%d, "
"fsync_after_written_bytes=%d, "
"sync_binlog_buff_interval=%ds, "
"sync_stat_file_interval=%ds, "
"upload_priority=%d, "
"if_alias_prefix=%s, "
"check_file_duplicate=%d, file_signature_method=%s, "
"FDHT group count=%d, FDHT server count=%d, "
"FDHT key_namespace=%s, FDHT keep_alive=%d, "
"HTTP server port=%d, domain name=%s, "
"use_access_log=%d, rotate_access_log=%d, "
"access_log_rotate_time=%02d:%02d, "
"compress_old_access_log=%d, "
"compress_access_log_days_before=%d, "
"rotate_access_log_size=%"PRId64", "
"file_sync_skip_invalid_record=%d, "
"use_connection_pool=%d, "
"g_connection_pool_max_idle_time=%ds, "
"compress_binlog=%d, "
"compress_binlog_time=%02d:%02d, "
"check_store_path_mark=%d",
g_fdfs_version.major, g_fdfs_version.minor,
sz_global_config, sz_service_config,
g_fdfs_store_paths.count, g_subdir_count_per_path,
g_group_name, g_client_bind_addr, g_disk_rw_separated,
g_disk_reader_threads, g_disk_writer_threads,
g_disk_recovery_threads, g_heart_beat_interval,
g_stat_report_interval, g_tracker_group.server_count,
g_sync_wait_usec / 1000, g_sync_interval / 1000,
g_sync_start_time.hour, g_sync_start_time.minute,
g_sync_end_time.hour, g_sync_end_time.minute,
g_write_mark_file_freq, g_allow_ip_count,
g_file_distribute_path_mode,
g_file_distribute_rotate_count,
g_fsync_after_written_bytes,
g_sync_binlog_buff_interval, g_sync_stat_file_interval,
g_upload_priority, g_if_alias_prefix, g_check_file_duplicate,
g_file_signature_method == STORAGE_FILE_SIGNATURE_METHOD_HASH
? "hash" : "md5",
g_group_array.group_count, g_group_array.server_count, \
g_key_namespace, g_keep_alive, \
g_http_port, g_http_domain, g_use_access_log, \
g_rotate_access_log, g_access_log_rotate_time.hour, \
g_access_log_rotate_time.minute, g_compress_old_access_log, \
g_compress_access_log_days_before, \
g_rotate_error_log, g_error_log_rotate_time.hour, \
g_error_log_rotate_time.minute, g_compress_old_error_log, \
g_compress_error_log_days_before, \
g_access_log_context.rotate_size, \
g_log_context.rotate_size, g_log_file_keep_days, \
g_file_sync_skip_invalid_record, \
g_use_connection_pool, g_connection_pool_max_idle_time, \
g_compress_binlog, g_compress_binlog_time.hour, \
g_group_array.group_count, g_group_array.server_count,
g_key_namespace, g_keep_alive,
g_http_port, g_http_domain, g_use_access_log,
g_rotate_access_log, g_access_log_rotate_time.hour,
g_access_log_rotate_time.minute, g_compress_old_access_log,
g_compress_access_log_days_before,
g_access_log_context.rotate_size,
g_file_sync_skip_invalid_record,
g_use_connection_pool, g_connection_pool_max_idle_time,
g_compress_binlog, g_compress_binlog_time.hour,
g_compress_binlog_time.minute, g_check_store_path_mark);
#ifdef WITH_HTTPD
@ -2401,7 +2198,7 @@ static int storage_get_my_ip_from_tracker(ConnectionInfo *conn,
pHeader->cmd = TRACKER_PROTO_CMD_STORAGE_GET_MY_IP;
strcpy(out_buff + sizeof(TrackerHeader), g_group_name);
if((result=tcpsenddata_nb(conn->sock, out_buff,
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, "
"tracker server %s:%d, send data fail, "
@ -2716,7 +2513,7 @@ int recv_file_serialized(int sock, const char *filename, \
}
if ((result=tcprecvdata_nb(sock, buff, recv_bytes, \
g_fdfs_network_timeout)) != 0)
SF_G_NETWORK_TIMEOUT)) != 0)
{
close(fd);
unlink(filename);

3
storage/storage_func.h
View File

@ -22,8 +22,7 @@ typedef char * (*get_filename_func)(const void *pArg, \
int storage_write_to_fd(int fd, get_filename_func filename_func, \
const void *pArg, const char *buff, const int len);
int storage_func_init(const char *filename, \
char *bind_addr, const int addr_size);
int storage_func_init(const char *filename);
int storage_func_destroy();
int storage_write_to_stat_file();

3
storage/storage_global.c
View File

@ -20,7 +20,6 @@ char g_http_domain[FDFS_DOMAIN_NAME_MAX_SIZE] = {0};
int g_http_port = 80;
int g_last_server_port = 0;
int g_last_http_port = 0;
int g_buff_size = STORAGE_DEFAULT_BUFF_SIZE;
bool g_disk_rw_direct = false;
bool g_disk_rw_separated = true;
@ -88,7 +87,6 @@ StorageStatusPerTracker *g_my_report_status = NULL; //returned by tracker serve
TimeInfo g_access_log_rotate_time = {0, 0}; //rotate access log time base
char SF_G_INNER_BIND_ADDR[IP_ADDRESS_SIZE] = {0};
bool g_client_bind_addr = true;
bool g_storage_ip_changed_auto_adjust = false;
bool g_thread_kill_done = false;
@ -110,7 +108,6 @@ char g_exe_name[256] = {0};
#endif
int g_compress_access_log_days_before = 0;
struct storage_nio_thread_data *g_nio_thread_data = NULL;
struct storage_dio_thread_data *g_dio_thread_data = NULL;
int storage_cmp_by_server_id(const void *p1, const void *p2)

22
storage/storage_global.h
View File

@ -16,11 +16,12 @@
#include <string.h>
#include <time.h>
#include "fastcommon/common_define.h"
#include "fastcommon/local_ip_func.h"
#include "fdfs_define.h"
#include "tracker_types.h"
#include "client_global.h"
#include "fdht_types.h"
#include "fastcommon/local_ip_func.h"
#include "storage_types.h"
#ifdef WITH_HTTPD
#include "fdfs_http_shared.h"
@ -44,21 +45,6 @@
extern "C" {
#endif
typedef struct
{
FDFSStorageBrief server;
int last_sync_src_timestamp;
} FDFSStorageServer;
typedef struct
{
signed char my_status; //my status from tracker server
signed char my_result; //my report result
signed char src_storage_result; //src storage report result
bool get_my_ip_done;
bool report_my_status;
} StorageStatusPerTracker;
/* subdirs under store path, g_subdir_count * g_subdir_count 2 level subdirs */
extern int g_subdir_count_per_path;
@ -66,7 +52,6 @@ extern int g_http_port; //http server port
extern int g_last_server_port;
extern int g_last_http_port; //last http server port
extern char g_http_domain[FDFS_DOMAIN_NAME_MAX_SIZE]; //http server domain name
extern int g_buff_size;
extern bool g_disk_rw_direct; //if file read / write directly
extern bool g_disk_rw_separated; //if disk read / write separated
@ -95,7 +80,7 @@ extern int g_sync_interval; //unit: milliseconds
extern TimeInfo g_sync_start_time;
extern TimeInfo g_sync_end_time;
extern bool g_sync_part_time; //true for part time, false for all time of a day
extern int g_sync_binlog_buff_interval; //sync binlog buff to disk every interval seconds
extern int g_sync_binlog_buff_interval;
extern int g_write_mark_file_freq; //write to mark file after sync N files
extern int g_sync_stat_file_interval; //sync storage stat info to disk interval
@ -158,7 +143,6 @@ extern char g_exe_name[256];
extern int g_compress_access_log_days_before;
extern struct storage_nio_thread_data *g_nio_thread_data; //network io thread data
extern struct storage_dio_thread_data *g_dio_thread_data; //disk io thread data
int storage_cmp_by_server_id(const void *p1, const void *p2);

4
storage/storage_ip_changed_dealer.c
View File

@ -45,7 +45,7 @@ static int storage_do_changelog_req(ConnectionInfo *pTrackerServer)
strcpy(out_buff + sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN,
g_my_server_id_str);
if((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \
@ -81,7 +81,7 @@ static int storage_report_ip_changed(ConnectionInfo *pTrackerServer)
IP_ADDRESS_SIZE, g_tracker_client_ip.ips[0].address);
if((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \

543
storage/storage_nio.c
View File

@ -1,543 +0,0 @@
/**
* Copyright (C) 2008 Happy Fish / YuQing
*
* FastDFS may be copied only under the terms of the GNU General
* Public License V3, which may be found in the FastDFS source kit.
* Please visit the FastDFS Home Page http://www.fastken.com/ for more detail.
**/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <string.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <errno.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <pthread.h>
#include "fastcommon/shared_func.h"
#include "fastcommon/sched_thread.h"
#include "fastcommon/logger.h"
#include "fastcommon/sockopt.h"
#include "fastcommon/fast_task_queue.h"
#include "tracker_types.h"
#include "tracker_proto.h"
#include "storage_global.h"
#include "storage_service.h"
#include "fastcommon/ioevent_loop.h"
#include "storage_dio.h"
#include "storage_nio.h"
static void client_sock_read(int sock, short event, void *arg);
static void client_sock_write(int sock, short event, void *arg);
static int storage_nio_init(struct fast_task_info *pTask);
void task_finish_clean_up(struct fast_task_info *pTask)
{
StorageClientInfo *pClientInfo;
pClientInfo = (StorageClientInfo *)pTask->arg;
if (pClientInfo->clean_func != NULL)
{
pClientInfo->clean_func(pTask);
}
ioevent_detach(&pTask->thread_data->ev_puller, pTask->event.fd);
close(pTask->event.fd);
pTask->event.fd = -1;
if (pTask->event.timer.expires > 0)
{
fast_timer_remove(&pTask->thread_data->timer,
&pTask->event.timer);
pTask->event.timer.expires = 0;
}
pTask->canceled = false;
memset(pTask->arg, 0, sizeof(StorageClientInfo));
free_queue_push(pTask);
__sync_fetch_and_sub(&g_storage_stat.connection.current_count, 1);
++g_stat_change_count;
}
static int set_recv_event(struct fast_task_info *pTask)
{
int result;
if (pTask->event.callback == client_sock_read)
{
return 0;
}
pTask->event.callback = client_sock_read;
if (ioevent_modify(&pTask->thread_data->ev_puller,
pTask->event.fd, IOEVENT_READ, pTask) != 0)
{
result = errno != 0 ? errno : ENOENT;
ioevent_add_to_deleted_list(pTask);
logError("file: "__FILE__", line: %d, "\
"ioevent_modify fail, " \
"errno: %d, error info: %s", \
__LINE__, result, STRERROR(result));
return result;
}
return 0;
}
static int set_send_event(struct fast_task_info *pTask)
{
int result;
if (pTask->event.callback == client_sock_write)
{
return 0;
}
pTask->event.callback = client_sock_write;
if (ioevent_modify(&pTask->thread_data->ev_puller,
pTask->event.fd, IOEVENT_WRITE, pTask) != 0)
{
result = errno != 0 ? errno : ENOENT;
ioevent_add_to_deleted_list(pTask);
logError("file: "__FILE__", line: %d, "\
"ioevent_modify fail, " \
"errno: %d, error info: %s", \
__LINE__, result, STRERROR(result));
return result;
}
return 0;
}
void storage_recv_notify_read(int sock, short event, void *arg)
{
struct fast_task_info *pTask;
StorageClientInfo *pClientInfo;
long task_addr;
int64_t remain_bytes;
int bytes;
int result;
while (1)
{
if ((bytes=read(sock, &task_addr, sizeof(task_addr))) < 0)
{
if (!(errno == EAGAIN || errno == EWOULDBLOCK))
{
logError("file: "__FILE__", line: %d, " \
"call read failed, " \
"errno: %d, error info: %s", \
__LINE__, errno, STRERROR(errno));
}
break;
}
else if (bytes == 0)
{
logError("file: "__FILE__", line: %d, " \
"call read failed, end of file", __LINE__);
break;
}
pTask = (struct fast_task_info *)task_addr;
pClientInfo = (StorageClientInfo *)pTask->arg;
if (pTask->event.fd < 0) //quit flag
{
return;
}
/* //logInfo("=====thread index: %d, pTask->event.fd=%d", \
pClientInfo->nio_thread_index, pTask->event.fd);
*/
if (pClientInfo->stage & FDFS_STORAGE_STAGE_DIO_THREAD)
{
pClientInfo->stage &= ~FDFS_STORAGE_STAGE_DIO_THREAD;
}
switch (pClientInfo->stage)
{
case FDFS_STORAGE_STAGE_NIO_INIT:
result = storage_nio_init(pTask);
break;
case FDFS_STORAGE_STAGE_NIO_RECV:
pTask->offset = 0;
remain_bytes = pClientInfo->total_length - \
pClientInfo->total_offset;
if (remain_bytes > pTask->size)
{
pTask->length = pTask->size;
}
else
{
pTask->length = remain_bytes;
}
if (set_recv_event(pTask) == 0)
{
client_sock_read(pTask->event.fd,
IOEVENT_READ, pTask);
}
result = 0;
break;
case FDFS_STORAGE_STAGE_NIO_SEND:
result = storage_send_add_event(pTask);
break;
case FDFS_STORAGE_STAGE_NIO_CLOSE:
result = EIO; //close this socket
break;
default:
logError("file: "__FILE__", line: %d, " \
"invalid stage: %d", __LINE__, \
pClientInfo->stage);
result = EINVAL;
break;
}
if (result != 0)
{
ioevent_add_to_deleted_list(pTask);
}
}
}
static int storage_nio_init(struct fast_task_info *pTask)
{
StorageClientInfo *pClientInfo;
struct storage_nio_thread_data *pThreadData;
pClientInfo = (StorageClientInfo *)pTask->arg;
pThreadData = g_nio_thread_data + pClientInfo->nio_thread_index;
pClientInfo->stage = FDFS_STORAGE_STAGE_NIO_RECV;
return ioevent_set(pTask, &pThreadData->thread_data,
pTask->event.fd, IOEVENT_READ, client_sock_read,
g_fdfs_network_timeout);
}
int storage_send_add_event(struct fast_task_info *pTask)
{
pTask->offset = 0;
/* direct send */
client_sock_write(pTask->event.fd, IOEVENT_WRITE, pTask);
return 0;
}
static void client_sock_read(int sock, short event, void *arg)
{
int bytes;
int recv_bytes;
struct fast_task_info *pTask;
StorageClientInfo *pClientInfo;
pTask = (struct fast_task_info *)arg;
pClientInfo = (StorageClientInfo *)pTask->arg;
if (pTask->canceled)
{
return;
}
if (pClientInfo->stage != FDFS_STORAGE_STAGE_NIO_RECV)
{
if (event & IOEVENT_TIMEOUT) {
pTask->event.timer.expires = g_current_time +
g_fdfs_network_timeout;
fast_timer_add(&pTask->thread_data->timer,
&pTask->event.timer);
}
return;
}
if (event & IOEVENT_TIMEOUT)
{
if (pClientInfo->total_offset == 0)
{
if (pTask->req_count > 0)
{
pTask->event.timer.expires = g_current_time +
g_fdfs_network_timeout;
fast_timer_add(&pTask->thread_data->timer,
&pTask->event.timer);
}
else
{
logWarning("file: "__FILE__", line: %d, "
"client ip: %s, recv timeout. "
"after the connection is established, "
"you must send a request before %ds timeout, "
"maybe connections leak in you application.",
__LINE__, pTask->client_ip, g_fdfs_network_timeout);
task_finish_clean_up(pTask);
}
}
else
{
logError("file: "__FILE__", line: %d, "
"client ip: %s, recv timeout, "
"recv offset: %d, expect length: %d, "
"req_count: %"PRId64, __LINE__, pTask->client_ip,
pTask->offset, pTask->length, pTask->req_count);
task_finish_clean_up(pTask);
}
return;
}
if (event & IOEVENT_ERROR)
{
logDebug("file: "__FILE__", line: %d, " \
"client ip: %s, recv error event: %d, "
"close connection", __LINE__, pTask->client_ip, event);
task_finish_clean_up(pTask);
return;
}
fast_timer_modify(&pTask->thread_data->timer,
&pTask->event.timer, g_current_time +
g_fdfs_network_timeout);
while (1)
{
if (pClientInfo->total_length == 0) //recv header
{
recv_bytes = sizeof(TrackerHeader) - pTask->offset;
}
else
{
recv_bytes = pTask->length - pTask->offset;
}
/*
logInfo("total_length=%"PRId64", recv_bytes=%d, "
"pTask->length=%d, pTask->offset=%d",
pClientInfo->total_length, recv_bytes,
pTask->length, pTask->offset);
*/
bytes = recv(sock, pTask->data + pTask->offset, recv_bytes, 0);
if (bytes < 0)
{
if (errno == EAGAIN || errno == EWOULDBLOCK)
{
}
else if (errno == EINTR)
{
continue;
}
else
{
logError("file: "__FILE__", line: %d, " \
"client ip: %s, recv failed, " \
"errno: %d, error info: %s", \
__LINE__, pTask->client_ip, \
errno, STRERROR(errno));
task_finish_clean_up(pTask);
}
return;
}
else if (bytes == 0)
{
logDebug("file: "__FILE__", line: %d, " \
"client ip: %s, recv failed, " \
"connection disconnected.", \
__LINE__, pTask->client_ip);
task_finish_clean_up(pTask);
return;
}
if (pClientInfo->total_length == 0) //header
{
if (pTask->offset + bytes < sizeof(TrackerHeader))
{
pTask->offset += bytes;
return;
}
pClientInfo->total_length=buff2long(((TrackerHeader *) \
pTask->data)->pkg_len);
if (pClientInfo->total_length < 0)
{
logError("file: "__FILE__", line: %d, " \
"client ip: %s, pkg length: " \
"%"PRId64" < 0", \
__LINE__, pTask->client_ip, \
pClientInfo->total_length);
task_finish_clean_up(pTask);
return;
}
pClientInfo->total_length += sizeof(TrackerHeader);
if (pClientInfo->total_length > pTask->size)
{
pTask->length = pTask->size;
}
else
{
pTask->length = pClientInfo->total_length;
}
}
pTask->offset += bytes;
if (pTask->offset >= pTask->length) //recv current pkg done
{
if (pClientInfo->total_offset + pTask->length >= \
pClientInfo->total_length)
{
/* current req recv done */
pClientInfo->stage = FDFS_STORAGE_STAGE_NIO_SEND;
pTask->req_count++;
}
if (pClientInfo->total_offset == 0)
{
pClientInfo->total_offset = pTask->length;
storage_deal_task(pTask);
}
else
{
pClientInfo->total_offset += pTask->length;
/* continue write to file */
storage_dio_queue_push(pTask);
}
return;
}
}
return;
}
static void client_sock_write(int sock, short event, void *arg)
{
int bytes;
struct fast_task_info *pTask;
StorageClientInfo *pClientInfo;
pTask = (struct fast_task_info *)arg;
pClientInfo = (StorageClientInfo *)pTask->arg;
if (pTask->canceled)
{
return;
}
if (event & IOEVENT_TIMEOUT)
{
logError("file: "__FILE__", line: %d, "
"client ip: %s, send timeout, offset: %d, "
"remain bytes: %d", __LINE__, pTask->client_ip,
pTask->offset, pTask->length - pTask->offset);
task_finish_clean_up(pTask);
return;
}
if (event & IOEVENT_ERROR)
{
logDebug("file: "__FILE__", line: %d, "
"client ip: %s, recv error event: %d, "
"close connection", __LINE__, pTask->client_ip, event);
task_finish_clean_up(pTask);
return;
}
while (1)
{
fast_timer_modify(&pTask->thread_data->timer,
&pTask->event.timer, g_current_time +
g_fdfs_network_timeout);
bytes = send(sock, pTask->data + pTask->offset, \
pTask->length - pTask->offset, 0);
//printf("%08X sended %d bytes\n", (int)pTask, bytes);
if (bytes < 0)
{
if (errno == EAGAIN || errno == EWOULDBLOCK)
{
set_send_event(pTask);
}
else if (errno == EINTR)
{
continue;
}
else
{
logError("file: "__FILE__", line: %d, " \
"client ip: %s, recv failed, " \
"errno: %d, error info: %s", \
__LINE__, pTask->client_ip, \
errno, STRERROR(errno));
task_finish_clean_up(pTask);
}
return;
}
else if (bytes == 0)
{
logWarning("file: "__FILE__", line: %d, " \
"send failed, connection disconnected.", \
__LINE__);
task_finish_clean_up(pTask);
return;
}
pTask->offset += bytes;
if (pTask->offset >= pTask->length)
{
if (set_recv_event(pTask) != 0)
{
return;
}
pClientInfo->total_offset += pTask->length;
if (pClientInfo->total_offset>=pClientInfo->total_length)
{
if (pClientInfo->total_length == sizeof(TrackerHeader)
&& ((TrackerHeader *)pTask->data)->status == EINVAL)
{
logDebug("file: "__FILE__", line: %d, "\
"close conn: #%d, client ip: %s", \
__LINE__, pTask->event.fd,
pTask->client_ip);
task_finish_clean_up(pTask);
return;
}
/* response done, try to recv again */
pClientInfo->total_length = 0;
pClientInfo->total_offset = 0;
pTask->offset = 0;
pTask->length = 0;
pClientInfo->stage = FDFS_STORAGE_STAGE_NIO_RECV;
}
else //continue to send file content
{
pTask->length = 0;
/* continue read from file */
storage_dio_queue_push(pTask);
}
return;
}
}
}

632
storage/storage_service.c
View File

@ -29,6 +29,8 @@
#include "fastcommon/pthread_func.h"
#include "fastcommon/sched_thread.h"
#include "fastcommon/fast_mblock.h"
#include "sf/sf_service.h"
#include "sf/sf_nio.h"
#include "tracker_types.h"
#include "tracker_proto.h"
#include "storage_service.h"
@ -41,7 +43,6 @@
#include "fdfs_global.h"
#include "tracker_client.h"
#include "storage_client.h"
#include "storage_nio.h"
#include "storage_dio.h"
#include "storage_sync.h"
#include "trunk_mem.h"
@ -68,19 +69,14 @@ typedef struct
int64_t fsize;
} StorageFileInfoForCRC32;
pthread_mutex_t g_storage_thread_lock;
int g_storage_thread_count = 0;
static int last_stat_change_count = 1; //for sync to stat file
static int64_t temp_file_sequence = 0;
static volatile int64_t temp_file_sequence = 0;
static pthread_mutex_t path_index_thread_lock;
static pthread_mutex_t stat_count_thread_lock;
static struct fast_mblock_man finfo_for_crc32_allocator;
static void *work_thread_entrance(void* arg);
extern int storage_client_create_link(ConnectionInfo *pTrackerServer, \
ConnectionInfo *pStorageServer, const char *master_filename,\
const char *src_filename, const int src_filename_len, \
@ -89,6 +85,8 @@ extern int storage_client_create_link(ConnectionInfo *pTrackerServer, \
const char *file_ext_name, \
char *remote_filename, int *filename_len);
static int storage_deal_task(struct fast_task_info *pTask, const int stage);
static int storage_do_delete_file(struct fast_task_info *pTask, \
DeleteFileLogCallback log_callback, \
FileDealDoneCallback done_callback, \
@ -107,7 +105,7 @@ static int storage_read_from_file(struct fast_task_info *pTask, \
static int storage_service_upload_file_done(struct fast_task_info *pTask);
#define STORAGE_STATUE_DEAL_FILE 123456 //status for read or write file
#define TASK_STATUS_CONTINUE 12345
#define FDHT_KEY_NAME_FILE_ID "fid"
#define FDHT_KEY_NAME_REF_COUNT "ref"
@ -315,7 +313,7 @@ static void storage_log_access_log(struct fast_task_info *pTask, \
#define STORAGE_ACCESS_LOG(pTask, action, status) \
do \
{ \
if (g_use_access_log && (status != STORAGE_STATUE_DEAL_FILE)) \
if (g_use_access_log && (status != TASK_STATUS_CONTINUE)) \
{ \
storage_log_access_log(pTask, action, status); \
} \
@ -446,7 +444,7 @@ static void storage_sync_delete_file_done_callback( \
long2buff(pClientInfo->total_length - sizeof(TrackerHeader), \
pHeader->pkg_len);
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
static void storage_sync_truncate_file_done_callback( \
@ -486,7 +484,7 @@ static void storage_sync_truncate_file_done_callback( \
long2buff(pClientInfo->total_length - sizeof(TrackerHeader), \
pHeader->pkg_len);
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
static int storage_sync_copy_file_rename_filename(
@ -594,7 +592,7 @@ static void storage_sync_copy_file_done_callback(struct fast_task_info *pTask, \
long2buff(pClientInfo->total_length - sizeof(TrackerHeader), \
pHeader->pkg_len);
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
static void storage_sync_modify_file_done_callback( \
@ -672,15 +670,9 @@ static void storage_sync_modify_file_done_callback( \
long2buff(pClientInfo->total_length - sizeof(TrackerHeader), \
pHeader->pkg_len);
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
#define STORAGE_NIO_NOTIFY_CLOSE(pTask) \
do { \
((StorageClientInfo *)pTask->arg)->stage = FDFS_STORAGE_STAGE_NIO_CLOSE; \
storage_nio_notify(pTask); \
} while (0)
static void storage_get_metadata_done_callback(struct fast_task_info *pTask, \
const int err_no)
{
@ -699,11 +691,11 @@ static void storage_get_metadata_done_callback(struct fast_task_info *pTask, \
{
pHeader = (TrackerHeader *)pTask->data;
pHeader->status = err_no;
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
else
{
STORAGE_NIO_NOTIFY_CLOSE(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_CLOSE);
}
}
else
@ -712,7 +704,7 @@ static void storage_get_metadata_done_callback(struct fast_task_info *pTask, \
g_storage_stat.total_get_meta_count, \
g_storage_stat.success_get_meta_count)
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
}
@ -738,11 +730,11 @@ static void storage_download_file_done_callback( \
{
pHeader = (TrackerHeader *)pTask->data;
pHeader->status = err_no;
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
else
{
STORAGE_NIO_NOTIFY_CLOSE(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_CLOSE);
}
}
else
@ -754,7 +746,7 @@ static void storage_download_file_done_callback( \
g_storage_stat.success_download_bytes, \
pFileContext->end - pFileContext->start)
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
}
@ -827,8 +819,7 @@ static int storage_do_delete_meta_file(struct fast_task_info *pTask)
src_file_nlink = -1;
if (g_check_file_duplicate)
{
pGroupArray=&((g_nio_thread_data+pClientInfo->nio_thread_index)\
->group_array);
pGroupArray = pTask->thread_data->arg;
memset(&key_info_sig, 0, sizeof(key_info_sig));
key_info_sig.namespace_len = g_namespace_len;
memcpy(key_info_sig.szNameSpace, g_key_namespace, \
@ -927,8 +918,7 @@ static int storage_do_delete_meta_file(struct fast_task_info *pTask)
struct stat stat_buf;
FDFSTrunkHeader trunkHeader;
pGroupArray=&((g_nio_thread_data+pClientInfo->nio_thread_index)\
->group_array);
pGroupArray = pTask->thread_data->arg;
if ((result=fdht_delete_ex(pGroupArray, g_keep_alive, \
&key_info_sig)) != 0)
{
@ -1123,7 +1113,7 @@ static void storage_delete_fdfs_file_done_callback( \
STORAGE_ACCESS_LOG(pTask, ACCESS_LOG_ACTION_DELETE_FILE, result);
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
static void storage_upload_file_done_callback(struct fast_task_info *pTask, \
@ -1216,7 +1206,7 @@ static void storage_upload_file_done_callback(struct fast_task_info *pTask, \
long2buff(pClientInfo->total_length - sizeof(TrackerHeader), \
pHeader->pkg_len);
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
static void storage_trunk_create_link_file_done_callback( \
@ -1306,7 +1296,7 @@ static void storage_trunk_create_link_file_done_callback( \
long2buff(pClientInfo->total_length - sizeof(TrackerHeader), \
pHeader->pkg_len);
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
}
@ -1377,7 +1367,7 @@ static void storage_append_file_done_callback(struct fast_task_info *pTask, \
STORAGE_ACCESS_LOG(pTask, ACCESS_LOG_ACTION_APPEND_FILE, result);
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
static void storage_modify_file_done_callback(struct fast_task_info *pTask, \
@ -1447,7 +1437,7 @@ static void storage_modify_file_done_callback(struct fast_task_info *pTask, \
STORAGE_ACCESS_LOG(pTask, ACCESS_LOG_ACTION_MODIFY_FILE, result);
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
static void storage_do_truncate_file_done_callback(struct fast_task_info *pTask, \
@ -1512,7 +1502,7 @@ static void storage_do_truncate_file_done_callback(struct fast_task_info *pTask,
STORAGE_ACCESS_LOG(pTask, ACCESS_LOG_ACTION_TRUNCATE_FILE, result);
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
static void storage_set_metadata_done_callback( \
@ -1568,25 +1558,66 @@ static void storage_set_metadata_done_callback( \
STORAGE_ACCESS_LOG(pTask, ACCESS_LOG_ACTION_SET_METADATA, result);
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
void task_finish_clean_up(struct fast_task_info *pTask)
{
StorageClientInfo *pClientInfo;
pClientInfo = (StorageClientInfo *)pTask->arg;
if (pClientInfo->clean_func != NULL)
{
pClientInfo->clean_func(pTask);
}
memset(pTask->arg, 0, sizeof(StorageClientInfo));
++g_stat_change_count;
sf_task_finish_clean_up(pTask);
}
static int sock_accept_done_callback(struct fast_task_info *task,
const in_addr_t client_addr, const bool bInnerPort)
{
if (g_allow_ip_count >= 0)
{
if (bsearch(&client_addr, g_allow_ip_addrs,
g_allow_ip_count, sizeof(in_addr_t),
cmp_by_ip_addr_t) == NULL)
{
logError("file: "__FILE__", line: %d, "
"ip addr %s is not allowed to access",
__LINE__, task->client_ip);
return EPERM;
}
}
return 0;
}
static void *alloc_thread_extra_data_func(const int thread_index)
{
int result;
GroupArray *group_array; //FastDHT group array
if (g_check_file_duplicate)
{
group_array = fc_malloc(sizeof(GroupArray));
if ((result=fdht_copy_group_array(group_array, &g_group_array)) != 0)
{
return NULL;
}
return group_array;
}
else
{
return NULL;
}
}
int storage_service_init()
{
#define ALLOC_CONNECTIONS_ONCE 256
int result;
int bytes;
int init_connections;
struct storage_nio_thread_data *pThreadData;
struct storage_nio_thread_data *pDataEnd;
pthread_t tid;
pthread_attr_t thread_attr;
if ((result=init_pthread_lock(&g_storage_thread_lock)) != 0)
{
return result;
}
if ((result=init_pthread_lock(&path_index_thread_lock)) != 0)
{
@ -1598,402 +1629,29 @@ int storage_service_init()
return result;
}
if ((result=init_pthread_attr(&thread_attr, SF_G_THREAD_STACK_SIZE)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"init_pthread_attr fail, program exit!", __LINE__);
return result;
}
init_connections = g_max_connections < ALLOC_CONNECTIONS_ONCE ?
g_max_connections : ALLOC_CONNECTIONS_ONCE;
if ((result=free_queue_init_ex(g_max_connections, init_connections,
ALLOC_CONNECTIONS_ONCE, g_buff_size,
g_buff_size, sizeof(StorageClientInfo))) != 0)
{
return result;
}
bytes = sizeof(struct storage_nio_thread_data) * g_work_threads;
g_nio_thread_data = (struct storage_nio_thread_data *)malloc(bytes);
if (g_nio_thread_data == NULL)
{
logError("file: "__FILE__", line: %d, " \
"malloc %d bytes fail, errno: %d, error info: %s", \
__LINE__, bytes, errno, STRERROR(errno));
return errno != 0 ? errno : ENOMEM;
}
memset(g_nio_thread_data, 0, bytes);
g_storage_thread_count = 0;
pDataEnd = g_nio_thread_data + g_work_threads;
for (pThreadData=g_nio_thread_data; pThreadData<pDataEnd; pThreadData++)
{
if (ioevent_init(&pThreadData->thread_data.ev_puller,
g_max_connections + 2, 1000, 0) != 0)
{
result = errno != 0 ? errno : ENOMEM;
logError("file: "__FILE__", line: %d, " \
"ioevent_init fail, " \
"errno: %d, error info: %s", \
__LINE__, result, STRERROR(result));
return result;
}
result = fast_timer_init(&pThreadData->thread_data.timer,
2 * g_fdfs_network_timeout, g_current_time);
if (result != 0)
{
logError("file: "__FILE__", line: %d, " \
"fast_timer_init fail, " \
"errno: %d, error info: %s", \
__LINE__, result, STRERROR(result));
return result;
}
if (pipe(pThreadData->thread_data.pipe_fds) != 0)
{
result = errno != 0 ? errno : EPERM;
logError("file: "__FILE__", line: %d, " \
"call pipe fail, " \
"errno: %d, error info: %s", \
__LINE__, result, STRERROR(result));
break;
}
#if defined(OS_LINUX)
if ((result=fd_add_flags(pThreadData->thread_data.pipe_fds[0], \
O_NONBLOCK | O_NOATIME)) != 0)
{
break;
}
#else
if ((result=fd_add_flags(pThreadData->thread_data.pipe_fds[0], \
O_NONBLOCK)) != 0)
{
break;
}
#endif
if ((result=pthread_create(&tid, &thread_attr, \
work_thread_entrance, pThreadData)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"create thread failed, startup threads: %d, " \
"errno: %d, error info: %s", \
__LINE__, g_storage_thread_count, \
result, STRERROR(result));
break;
}
else
{
if ((result=pthread_mutex_lock(&g_storage_thread_lock)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"call pthread_mutex_lock fail, " \
"errno: %d, error info: %s", \
__LINE__, result, STRERROR(result));
}
g_storage_thread_count++;
if ((result=pthread_mutex_unlock(&g_storage_thread_lock)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"call pthread_mutex_lock fail, " \
"errno: %d, error info: %s", \
__LINE__, result, STRERROR(result));
}
}
}
pthread_attr_destroy(&thread_attr);
last_stat_change_count = g_stat_change_count;
//DO NOT support direct IO !!!
//g_extra_open_file_flags = g_disk_rw_direct ? O_DIRECT : 0;
if (result != 0)
if ((result=fast_mblock_init(&finfo_for_crc32_allocator,
sizeof(StorageFileInfoForCRC32), 1024)) != 0)
{
return result;
}
result = fast_mblock_init(&finfo_for_crc32_allocator,
sizeof(StorageFileInfoForCRC32), 1024);
result = sf_service_init("fdfs_storaged", alloc_thread_extra_data_func,
NULL, sock_accept_done_callback, fdfs_set_body_length,
storage_deal_task, task_finish_clean_up, NULL, 1000,
sizeof(TrackerHeader), sizeof(StorageClientInfo));
sf_enable_thread_notify(false);
sf_set_remove_from_ready_list(false);
return result;
}
void storage_service_destroy()
{
pthread_mutex_destroy(&g_storage_thread_lock);
pthread_mutex_destroy(&path_index_thread_lock);
pthread_mutex_destroy(&stat_count_thread_lock);
}
int storage_terminate_threads()
{
struct storage_nio_thread_data *pThreadData;
struct storage_nio_thread_data *pDataEnd;
struct fast_task_info *pTask;
StorageClientInfo *pClientInfo;
long task_addr;
int quit_sock;
if (g_nio_thread_data != NULL)
{
pDataEnd = g_nio_thread_data + g_work_threads;
quit_sock = 0;
for (pThreadData=g_nio_thread_data; pThreadData<pDataEnd; \
pThreadData++)
{
quit_sock--;
pTask = free_queue_pop();
if (pTask == NULL)
{
logError("file: "__FILE__", line: %d, " \
"malloc task buff failed, you should " \
"increase the parameter: max_connections",
__LINE__);
continue;
}
pClientInfo = (StorageClientInfo *)pTask->arg;
pTask->event.fd = quit_sock;
pClientInfo->nio_thread_index = pThreadData - g_nio_thread_data;
task_addr = (long)pTask;
if (write(pThreadData->thread_data.pipe_fds[1], &task_addr, \
sizeof(task_addr)) != sizeof(task_addr))
{
logError("file: "__FILE__", line: %d, " \
"call write failed, " \
"errno: %d, error info: %s", \
__LINE__, errno, STRERROR(errno));
}
}
}
return 0;
}
static void *accept_thread_entrance(void* arg)
{
int server_sock;
int incomesock;
struct sockaddr_in inaddr;
socklen_t sockaddr_len;
in_addr_t client_addr;
char szClientIp[IP_ADDRESS_SIZE];
long task_addr;
struct fast_task_info *pTask;
StorageClientInfo *pClientInfo;
struct storage_nio_thread_data *pThreadData;
server_sock = (long)arg;
while (SF_G_CONTINUE_FLAG)
{
sockaddr_len = sizeof(inaddr);
incomesock = accept(server_sock, (struct sockaddr*)&inaddr, \
&sockaddr_len);
if (incomesock < 0) //error
{
if (!(errno == EINTR || errno == EAGAIN))
{
logError("file: "__FILE__", line: %d, " \
"accept failed, " \
"errno: %d, error info: %s", \
__LINE__, errno, STRERROR(errno));
}
continue;
}
client_addr = getPeerIpaddr(incomesock, \
szClientIp, IP_ADDRESS_SIZE);
if (g_allow_ip_count >= 0)
{
if (bsearch(&client_addr, g_allow_ip_addrs, \
g_allow_ip_count, sizeof(in_addr_t), \
cmp_by_ip_addr_t) == NULL)
{
logError("file: "__FILE__", line: %d, " \
"ip addr %s is not allowed to access", \
__LINE__, szClientIp);
close(incomesock);
continue;
}
}
if (tcpsetnonblockopt(incomesock) != 0)
{
close(incomesock);
continue;
}
pTask = free_queue_pop();
if (pTask == NULL)
{
logError("file: "__FILE__", line: %d, "
"malloc task buff fail, you should "
"increase the parameter \"max_connections\" "
"in storage.conf, or check your applications "
"for connection leaks", __LINE__);
close(incomesock);
continue;
}
pClientInfo = (StorageClientInfo *)pTask->arg;
pTask->event.fd = incomesock;
pClientInfo->stage = FDFS_STORAGE_STAGE_NIO_INIT;
pClientInfo->nio_thread_index = pTask->event.fd % g_work_threads;
pThreadData = g_nio_thread_data + pClientInfo->nio_thread_index;
strcpy(pTask->client_ip, szClientIp);
task_addr = (long)pTask;
if (write(pThreadData->thread_data.pipe_fds[1], &task_addr, \
sizeof(task_addr)) != sizeof(task_addr))
{
close(incomesock);
free_queue_push(pTask);
logError("file: "__FILE__", line: %d, " \
"call write failed, " \
"errno: %d, error info: %s", \
__LINE__, errno, STRERROR(errno));
}
else
{
int current_connections;
current_connections = __sync_add_and_fetch(&g_storage_stat.connection.
current_count, 1);
if (current_connections > g_storage_stat.connection.max_count) {
g_storage_stat.connection.max_count = current_connections;
}
++g_stat_change_count;
}
}
return NULL;
}
void storage_accept_loop(int server_sock)
{
if (g_accept_threads > 1)
{
pthread_t tid;
pthread_attr_t thread_attr;
int result;
int i;
if ((result=init_pthread_attr(&thread_attr, SF_G_THREAD_STACK_SIZE)) != 0)
{
logWarning("file: "__FILE__", line: %d, " \
"init_pthread_attr fail!", __LINE__);
}
else
{
for (i=1; i<g_accept_threads; i++)
{
if ((result=pthread_create(&tid, &thread_attr, \
accept_thread_entrance, \
(void *)(long)server_sock)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"create thread failed, startup threads: %d, " \
"errno: %d, error info: %s", \
__LINE__, i, result, STRERROR(result));
break;
}
}
pthread_attr_destroy(&thread_attr);
}
}
accept_thread_entrance((void *)(long)server_sock);
}
int storage_nio_notify(struct fast_task_info *pTask)
{
StorageClientInfo *pClientInfo;
struct storage_nio_thread_data *pThreadData;
long task_addr;
pClientInfo = (StorageClientInfo *)pTask->arg;
pThreadData = g_nio_thread_data + pClientInfo->nio_thread_index;
task_addr = (long)pTask;
if (write(pThreadData->thread_data.pipe_fds[1], &task_addr, \
sizeof(task_addr)) != sizeof(task_addr))
{
int result;
result = errno != 0 ? errno : EIO;
logCrit("file: "__FILE__", line: %d, " \
"call write failed, " \
"errno: %d, error info: %s", \
__LINE__, result, STRERROR(result));
abort();
}
return 0;
}
static void *work_thread_entrance(void* arg)
{
int result;
struct storage_nio_thread_data *pThreadData;
pThreadData = (struct storage_nio_thread_data *)arg;
if (g_check_file_duplicate)
{
if ((result=fdht_copy_group_array(&(pThreadData->group_array),\
&g_group_array)) != 0)
{
pthread_mutex_lock(&g_storage_thread_lock);
g_storage_thread_count--;
pthread_mutex_unlock(&g_storage_thread_lock);
return NULL;
}
}
ioevent_loop(&pThreadData->thread_data, storage_recv_notify_read,
task_finish_clean_up, &SF_G_CONTINUE_FLAG);
ioevent_destroy(&pThreadData->thread_data.ev_puller);
if (g_check_file_duplicate)
{
if (g_keep_alive)
{
fdht_disconnect_all_servers(&(pThreadData->group_array));
}
fdht_free_group_array(&(pThreadData->group_array));
}
if ((result=pthread_mutex_lock(&g_storage_thread_lock)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"call pthread_mutex_lock fail, " \
"errno: %d, error info: %s", \
__LINE__, result, STRERROR(result));
}
g_storage_thread_count--;
if ((result=pthread_mutex_unlock(&g_storage_thread_lock)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"call pthread_mutex_lock fail, " \
"errno: %d, error info: %s", \
__LINE__, result, STRERROR(result));
}
logDebug("file: "__FILE__", line: %d, " \
"nio thread exited, thread count: %d", \
__LINE__, g_storage_thread_count);
return NULL;
}
int storage_get_storage_path_index(int *store_path_index)
{
int i;
@ -2352,7 +2010,7 @@ static int storage_client_create_link_wrapper(struct fast_task_info *pTask, \
master_filename, strlen(master_filename), \
prefix_name, file_ext_name, \
remote_filename, filename_len, false);
if (result == STORAGE_STATUE_DEAL_FILE)
if (result == TASK_STATUS_CONTINUE)
{
result = 0;
}
@ -2592,9 +2250,7 @@ static int storage_service_upload_file_done(struct fast_task_info *pTask)
key_info.namespace_len = g_namespace_len;
memcpy(key_info.szNameSpace, g_key_namespace, g_namespace_len);
pGroupArray=&((g_nio_thread_data+pClientInfo->nio_thread_index)\
->group_array);
pGroupArray = pTask->thread_data->arg;
STORAGE_GEN_FILE_SIGNATURE(file_size, \
pFileContext->file_hash_codes, szFileSig)
/*
@ -2759,6 +2415,37 @@ static int storage_service_upload_file_done(struct fast_task_info *pTask)
return 0;
}
static int storage_nio_notify(struct fast_task_info *pTask, const int stage)
{
StorageClientInfo *pClientInfo;
int64_t remain_bytes;
pClientInfo = (StorageClientInfo *)pTask->arg;
if (stage == SF_NIO_STAGE_RECV)
{
pTask->offset = 0;
remain_bytes = pClientInfo->total_length -
pClientInfo->total_offset;
if (remain_bytes > pTask->size)
{
pTask->length = pTask->size;
}
else
{
pTask->length = remain_bytes;
}
}
return sf_nio_notify(pTask, stage);
}
static int calc_crc32_continue_callback(struct fast_task_info *pTask,
const int stage)
{
pTask->length = 0;
return storage_dio_queue_push(pTask);
}
static int storage_trunk_do_create_link(struct fast_task_info *pTask, \
const int64_t file_bytes, const int buff_offset, \
FileBeforeOpenCallback before_open_callback,
@ -2843,7 +2530,7 @@ static int storage_trunk_create_link(struct fast_task_info *pTask, \
storage_trunk_do_create_link(pTask, file_bytes, p - pTask->data, \
dio_check_trunk_file_when_upload, \
storage_trunk_create_link_file_done_callback);
return STORAGE_STATUE_DEAL_FILE;
return TASK_STATUS_CONTINUE;
}
static int storage_service_do_create_link(struct fast_task_info *pTask, \
@ -2932,9 +2619,7 @@ static int storage_set_link_file_meta(struct fast_task_info *pTask, \
key_info.namespace_len = g_namespace_len;
memcpy(key_info.szNameSpace, g_key_namespace, g_namespace_len);
pGroupArray=&((g_nio_thread_data + pClientInfo->nio_thread_index) \
->group_array);
pGroupArray = pTask->thread_data->arg;
key_info.obj_id_len = snprintf(key_info.szObjectId, \
sizeof(key_info.szObjectId), \
"%s/%c"FDFS_STORAGE_DATA_DIR_FORMAT"/%s", \
@ -3356,7 +3041,7 @@ static int storage_server_set_metadata(struct fast_task_info *pTask)
return result;
}
return STORAGE_STATUE_DEAL_FILE;
return TASK_STATUS_CONTINUE;
}
/**
@ -3527,13 +3212,7 @@ static void calc_crc32_done_callback_for_query_finfo(
long2buff(pTask->length - sizeof(TrackerHeader), pHeader->pkg_len);
STORAGE_ACCESS_LOG(pTask, ACCESS_LOG_ACTION_QUERY_FILE, result);
storage_nio_notify(pTask);
}
static int calc_crc32_continue_callback(struct fast_task_info *pTask)
{
pTask->length = 0;
return storage_dio_queue_push(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
static int push_calc_crc32_to_dio_queue(struct fast_task_info *pTask,
@ -4172,7 +3851,7 @@ static int storage_server_fetch_one_path_binlog_dealer(
if (pClientInfo->total_length - pClientInfo->total_offset <=
STORAGE_LAST_AHEAD_BYTES) //finished, close the connection
{
STORAGE_NIO_NOTIFY_CLOSE(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_CLOSE);
return 0;
}
@ -4371,7 +4050,7 @@ static int storage_server_fetch_one_path_binlog_dealer(
if (result != 0) //error occurs
{
STORAGE_NIO_NOTIFY_CLOSE(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_CLOSE);
return result;
}
@ -4387,7 +4066,7 @@ static int storage_server_fetch_one_path_binlog_dealer(
+ STORAGE_LAST_AHEAD_BYTES;
}
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
return 0;
}
@ -4478,9 +4157,8 @@ static int storage_server_do_fetch_one_path_binlog(
long2buff(pClientInfo->total_length - sizeof(TrackerHeader),
pHeader->pkg_len);
storage_nio_notify(pTask);
return STORAGE_STATUE_DEAL_FILE;
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
return TASK_STATUS_CONTINUE;
}
/**
@ -4944,7 +4622,7 @@ static void calc_crc32_done_callback_for_regenerate(
long2buff(pTask->length - sizeof(TrackerHeader), pHeader->pkg_len);
STORAGE_ACCESS_LOG(pTask, ACCESS_LOG_ACTION_RENAME_FILE, result);
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
}
/**
@ -5887,15 +5565,9 @@ static int storage_sync_copy_file(struct fast_task_info *pTask, \
for (i=0; i < MKTEMP_MAX_COUNT; i++)
{
pthread_mutex_lock(&g_storage_thread_lock);
sprintf(pFileContext->filename, "%s/data/.cp" \
"%"PRId64".tmp", \
g_fdfs_store_paths.paths[store_path_index].path, \
temp_file_sequence++);
pthread_mutex_unlock(&g_storage_thread_lock);
sprintf(pFileContext->filename, "%s/data/.cp%"PRId64".tmp",
g_fdfs_store_paths.paths[store_path_index].path,
__sync_add_and_fetch(&temp_file_sequence, 1));
if (stat(pFileContext->filename, &stat_buf) == 0)
{
if (g_current_time - stat_buf.st_mtime > 600)
@ -5968,7 +5640,7 @@ static int storage_sync_copy_file(struct fast_task_info *pTask, \
else
{
storage_sync_copy_file_done_callback(pTask, 0);
return STORAGE_STATUE_DEAL_FILE;
return TASK_STATUS_CONTINUE;
}
}
@ -6801,8 +6473,7 @@ static int storage_do_sync_link_file(struct fast_task_info *pTask)
long2buff(pClientInfo->total_length - sizeof(TrackerHeader), \
pHeader->pkg_len);
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
return result;
}
@ -6901,7 +6572,7 @@ static int storage_sync_link_file(struct fast_task_info *pTask)
return result;
}
return STORAGE_STATUE_DEAL_FILE;
return TASK_STATUS_CONTINUE;
}
static int storage_sync_rename_file(struct fast_task_info *pTask)
@ -7361,7 +7032,7 @@ static int storage_do_delete_file(struct fast_task_info *pTask, \
return result;
}
return STORAGE_STATUE_DEAL_FILE;
return TASK_STATUS_CONTINUE;
}
static int storage_read_from_file(struct fast_task_info *pTask, \
@ -7411,7 +7082,7 @@ static int storage_read_from_file(struct fast_task_info *pTask, \
return result;
}
return STORAGE_STATUE_DEAL_FILE;
return TASK_STATUS_CONTINUE;
}
static int storage_write_to_file(struct fast_task_info *pTask, \
@ -7461,7 +7132,7 @@ static int storage_write_to_file(struct fast_task_info *pTask, \
return result;
}
return STORAGE_STATUE_DEAL_FILE;
return TASK_STATUS_CONTINUE;
}
/**
@ -7855,8 +7526,7 @@ static int storage_create_link_core(struct fast_task_info *pTask, \
src_filename, result, STRERROR(result));
if (g_check_file_duplicate)
{
pGroupArray=&((g_nio_thread_data+pClientInfo->nio_thread_index)\
->group_array);
pGroupArray = pTask->thread_data->arg;
//clean invalid entry
memset(&key_info, 0, sizeof(key_info));
key_info.namespace_len = g_namespace_len;
@ -8185,7 +7855,7 @@ static int storage_do_create_link(struct fast_task_info *pTask)
master_filename, master_filename_len, \
prefix_name, file_ext_name, \
filename, &filename_len, true);
if (result == STORAGE_STATUE_DEAL_FILE)
if (result == TASK_STATUS_CONTINUE)
{
return 0;
}
@ -8207,8 +7877,7 @@ static int storage_do_create_link(struct fast_task_info *pTask)
long2buff(pClientInfo->total_length - sizeof(TrackerHeader), \
pHeader->pkg_len);
storage_nio_notify(pTask);
sf_nio_notify(pTask, SF_NIO_STAGE_SEND);
return result;
}
@ -8279,7 +7948,7 @@ static int storage_create_link(struct fast_task_info *pTask)
return result;
}
return STORAGE_STATUE_DEAL_FILE;
return TASK_STATUS_CONTINUE;
}
int fdfs_stat_file_sync_func(void *args)
@ -8315,7 +7984,7 @@ int fdfs_stat_file_sync_func(void *args)
} \
} while (0)
int storage_deal_task(struct fast_task_info *pTask)
static int storage_deal_task(struct fast_task_info *pTask, const int stage)
{
TrackerHeader *pHeader;
StorageClientInfo *pClientInfo;
@ -8479,7 +8148,10 @@ int storage_deal_task(struct fast_task_info *pTask)
break;
}
if (result != STORAGE_STATUE_DEAL_FILE)
if (result == TASK_STATUS_CONTINUE) {
return 0;
}
else
{
pClientInfo->total_offset = 0;
if (result != 0)
@ -8493,9 +8165,7 @@ int storage_deal_task(struct fast_task_info *pTask)
pHeader->cmd = STORAGE_PROTO_CMD_RESP;
long2buff(pClientInfo->total_length - sizeof(TrackerHeader), \
pHeader->pkg_len);
storage_send_add_event(pTask);
}
sf_send_add_event(pTask);
return result;
}
}

10
storage/storage_service.h
View File

@ -27,22 +27,14 @@
extern "C" {
#endif
extern int g_storage_thread_count;
extern pthread_mutex_t g_storage_thread_lock;
int storage_service_init();
void storage_service_destroy();
int fdfs_stat_file_sync_func(void *args);
int storage_deal_task(struct fast_task_info *pTask);
int storage_nio_notify(struct fast_task_info *pTask);
void storage_accept_loop(int server_sock);
int storage_terminate_threads();
int storage_get_storage_path_index(int *store_path_index);
void storage_get_store_path(const char *filename, const int filename_len, \
void storage_get_store_path(const char *filename, const int filename_len,
int *sub_path_high, int *sub_path_low);
#ifdef __cplusplus

30
storage/storage_sync.c
View File

@ -227,7 +227,7 @@ static int storage_sync_copy_file(ConnectionInfo *pStorageServer, \
p += pRecord->filename_len;
if((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
p - out_buff, g_fdfs_network_timeout)) != 0)
p - out_buff, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"sync data to storage server %s:%d fail, " \
@ -242,7 +242,7 @@ static int storage_sync_copy_file(ConnectionInfo *pStorageServer, \
if (need_sync_file && (stat_buf.st_size > 0) && \
((result=tcpsendfile_ex(pStorageServer->sock, \
full_filename, file_offset, stat_buf.st_size, \
g_fdfs_network_timeout, &total_send_bytes)) != 0))
SF_G_NETWORK_TIMEOUT, &total_send_bytes)) != 0))
{
logError("file: "__FILE__", line: %d, " \
"sync data to storage server %s:%d fail, " \
@ -424,7 +424,7 @@ static int storage_sync_modify_file(ConnectionInfo *pStorageServer, \
p += pRecord->filename_len;
if((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
p - out_buff, g_fdfs_network_timeout)) != 0)
p - out_buff, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"sync data to storage server %s:%d fail, " \
@ -438,7 +438,7 @@ static int storage_sync_modify_file(ConnectionInfo *pStorageServer, \
if ((result=tcpsendfile_ex(pStorageServer->sock, \
full_filename, start_offset, modify_length, \
g_fdfs_network_timeout, &total_send_bytes)) != 0)
SF_G_NETWORK_TIMEOUT, &total_send_bytes)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"sync data to storage server %s:%d fail, " \
@ -589,7 +589,7 @@ static int storage_sync_truncate_file(ConnectionInfo *pStorageServer, \
p += pRecord->filename_len;
if((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
p - out_buff, g_fdfs_network_timeout)) != 0)
p - out_buff, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"sync data to storage server %s:%d fail, " \
@ -663,7 +663,7 @@ static int storage_sync_delete_file(ConnectionInfo *pStorageServer, \
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
sizeof(TrackerHeader) + 4 + FDFS_GROUP_NAME_MAX_LEN + \
pRecord->filename_len, g_fdfs_network_timeout)) != 0)
pRecord->filename_len, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("FILE: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -713,7 +713,7 @@ static int storage_report_my_server_id(ConnectionInfo *pStorageServer)
strcpy(out_buff + sizeof(TrackerHeader), g_my_server_id_str);
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
sizeof(TrackerHeader) + FDFS_STORAGE_ID_MAX_SIZE, \
g_fdfs_network_timeout)) != 0)
SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("FILE: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -938,7 +938,7 @@ static int storage_sync_link_file(ConnectionInfo *pStorageServer, \
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
sizeof(TrackerHeader) + out_body_len, \
g_fdfs_network_timeout)) != 0)
SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("FILE: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -1046,7 +1046,7 @@ static int storage_sync_rename_file(ConnectionInfo *pStorageServer,
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff,
sizeof(TrackerHeader) + out_body_len,
g_fdfs_network_timeout)) != 0)
SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("FILE: "__FILE__", line: %d, "
"send data to storage server %s:%d fail, "
@ -1245,7 +1245,7 @@ static int write_to_binlog_index(const int binlog_index)
close(fd);
STORAGE_CHOWN(full_filename, geteuid(), getegid())
SF_CHOWN_TO_RUNBY_RETURN_ON_ERROR(full_filename);
return 0;
}
@ -1400,7 +1400,7 @@ static int open_next_writable_binlog()
errno, STRERROR(errno));
return errno != 0 ? errno : EACCES;
}
STORAGE_FCHOWN(g_binlog_fd, full_filename, geteuid(), getegid())
SF_FCHOWN_TO_RUNBY_RETURN_ON_ERROR(g_binlog_fd, full_filename);
g_binlog_index++;
return 0;
@ -1426,7 +1426,7 @@ int storage_sync_init()
return errno != 0 ? errno : ENOENT;
}
STORAGE_CHOWN(data_path, geteuid(), getegid())
SF_CHOWN_TO_RUNBY_RETURN_ON_ERROR(data_path);
}
snprintf(sync_path, sizeof(sync_path), \
@ -1443,7 +1443,7 @@ int storage_sync_init()
return errno != 0 ? errno : ENOENT;
}
STORAGE_CHOWN(sync_path, geteuid(), getegid())
SF_CHOWN_TO_RUNBY_RETURN_ON_ERROR(sync_path);
}
binlog_write_cache_buff = (char *)malloc(SYNC_BINLOG_WRITE_BUFF_SIZE);
@ -1486,7 +1486,7 @@ int storage_sync_init()
return errno != 0 ? errno : EIO;
}
STORAGE_FCHOWN(g_binlog_fd, full_filename, geteuid(), getegid())
SF_FCHOWN_TO_RUNBY_RETURN_ON_ERROR(g_binlog_fd, full_filename);
/*
//printf("full_filename=%s, binlog_file_size=%d\n", \
@ -2437,7 +2437,7 @@ static int storage_write_to_mark_file(StorageBinLogReader *pReader)
if ((result=safeWriteToFile(pReader->mark_filename, buff, len)) == 0)
{
STORAGE_CHOWN(pReader->mark_filename, geteuid(), getegid())
SF_CHOWN_TO_RUNBY_RETURN_ON_ERROR(pReader->mark_filename);
pReader->last_scan_rows = pReader->scan_row_count;
pReader->last_sync_rows = pReader->sync_row_count;
}

2
storage/storage_sync_func.c
View File

@ -96,7 +96,7 @@ void storage_sync_connect_storage_server_ex(const FDFSStorageBrief *pStorage,
if ((conn_results[i]=connectserverbyip_nb(conn->sock,
conn->ip_addr, SF_G_INNER_PORT,
g_fdfs_connect_timeout)) == 0)
SF_G_CONNECT_TIMEOUT)) == 0)
{
char szFailPrompt[64];
if (nContinuousFail == 0)

60
storage/storage_nio.h → storage/storage_types.h
View File

@ -6,29 +6,21 @@
* Please visit the FastDFS Home Page http://www.fastken.com/ for more detail.
**/
//tracker_nio.h
//storage_types.h
#ifndef _TRACKER_NIO_H
#define _TRACKER_NIO_H
#ifndef _STORAGE_TYPES_H
#define _STORAGE_TYPES_H
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include "tracker_types.h"
#include "storage_func.h"
#include "fastcommon/fast_task_queue.h"
#include "storage_global.h"
#include "tracker_types.h"
#include "fdht_types.h"
#include "trunk_mem.h"
#include "fastcommon/md5.h"
#define FDFS_STORAGE_STAGE_NIO_INIT 0
#define FDFS_STORAGE_STAGE_NIO_RECV 1
#define FDFS_STORAGE_STAGE_NIO_SEND 2
#define FDFS_STORAGE_STAGE_NIO_CLOSE 4 //close socket
#define FDFS_STORAGE_STAGE_DIO_THREAD 8
#define FDFS_STORAGE_FILE_OP_READ 'R'
#define FDFS_STORAGE_FILE_OP_WRITE 'W'
#define FDFS_STORAGE_FILE_OP_APPEND 'A'
@ -40,13 +32,14 @@ typedef int (*TaskDealFunc)(struct fast_task_info *pTask);
/* this clean func will be called when connection disconnected */
typedef void (*DisconnectCleanFunc)(struct fast_task_info *pTask);
typedef void (*DeleteFileLogCallback)(struct fast_task_info *pTask, \
typedef void (*DeleteFileLogCallback)(struct fast_task_info *pTask,
const int err_no);
typedef void (*FileDealDoneCallback)(struct fast_task_info *pTask, \
typedef void (*FileDealDoneCallback)(struct fast_task_info *pTask,
const int err_no);
typedef int (*FileDealContinueCallback)(struct fast_task_info *pTask);
typedef int (*FileDealContinueCallback)(struct fast_task_info *pTask,
const int stage);
typedef int (*FileBeforeOpenCallback)(struct fast_task_info *pTask);
typedef int (*FileBeforeCloseCallback)(struct fast_task_info *pTask);
@ -57,6 +50,21 @@ typedef int (*FileBeforeCloseCallback)(struct fast_task_info *pTask);
#define _FILE_TYPE_REGULAR 8
#define _FILE_TYPE_LINK 16
typedef struct
{
FDFSStorageBrief server;
int last_sync_src_timestamp;
} FDFSStorageServer;
typedef struct
{
signed char my_status; //my status from tracker server
signed char my_result; //my report result
signed char src_storage_result; //src storage report result
bool get_my_ip_done;
bool report_my_status;
} StorageStatusPerTracker;
typedef struct
{
bool if_gen_filename; //if upload generate filename
@ -120,8 +128,6 @@ typedef struct
typedef struct
{
int nio_thread_index; //nio thread index
char stage; //nio stage, send or recv
char storage_server_id[FDFS_STORAGE_ID_MAX_SIZE];
StorageFileContext file_context;
@ -137,24 +143,4 @@ typedef struct
DisconnectCleanFunc clean_func; //clean function pointer when finished
} StorageClientInfo;
struct storage_nio_thread_data
{
struct nio_thread_data thread_data;
GroupArray group_array; //FastDHT group array
};
#ifdef __cplusplus
extern "C" {
#endif
void storage_recv_notify_read(int sock, short event, void *arg);
int storage_send_add_event(struct fast_task_info *pTask);
void task_finish_clean_up(struct fast_task_info *pTask);
#ifdef __cplusplus
}
#endif
#endif

44
storage/tracker_client_thread.c
View File

@ -282,7 +282,7 @@ static void *tracker_report_thread_entrance(void *arg)
break;
}
tcpsetserveropt(conn->sock, g_fdfs_network_timeout);
tcpsetserveropt(conn->sock, SF_G_NETWORK_TIMEOUT);
getSockIpaddr(conn->sock, tracker_client_ip, IP_ADDRESS_SIZE);
if (nContinuousFail == 0)
@ -611,7 +611,7 @@ int tracker_sync_diff_servers(ConnectionInfo *pTrackerServer, \
out_len = sizeof(FDFSStorageBrief) * server_count;
long2buff(out_len, resp.pkg_len);
if ((result=tcpsenddata_nb(pTrackerServer->sock, &resp, sizeof(resp), \
g_fdfs_network_timeout)) != 0)
SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"trackert server %s:%d, send data fail, " \
@ -622,7 +622,7 @@ int tracker_sync_diff_servers(ConnectionInfo *pTrackerServer, \
}
if ((result=tcpsenddata_nb(pTrackerServer->sock, \
briefServers, out_len, g_fdfs_network_timeout)) != 0)
briefServers, out_len, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"trackert server %s:%d, send data fail, " \
@ -634,7 +634,7 @@ int tracker_sync_diff_servers(ConnectionInfo *pTrackerServer, \
if ((result=tcprecvdata_nb(pTrackerServer->sock, &resp, \
sizeof(resp), g_fdfs_network_timeout)) != 0)
sizeof(resp), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, recv data fail, " \
@ -678,7 +678,7 @@ int tracker_report_storage_status(ConnectionInfo *pTrackerServer, \
briefServer, sizeof(FDFSStorageBrief));
if ((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN + \
sizeof(FDFSStorageBrief), g_fdfs_network_timeout)) != 0)
sizeof(FDFSStorageBrief), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"trackert server %s:%d, send data fail, " \
@ -689,7 +689,7 @@ int tracker_report_storage_status(ConnectionInfo *pTrackerServer, \
}
if ((result=tcprecvdata_nb(pTrackerServer->sock, &resp, \
sizeof(resp), g_fdfs_network_timeout)) != 0)
sizeof(resp), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, recv data fail, " \
@ -1050,7 +1050,7 @@ static int _notify_reselect_tleader(ConnectionInfo *conn)
memset(out_buff, 0, sizeof(out_buff));
pHeader->cmd = TRACKER_PROTO_CMD_TRACKER_NOTIFY_RESELECT_LEADER;
if ((result=tcpsenddata_nb(conn->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, "
"tracker server %s:%d, send data fail, "
@ -1289,7 +1289,7 @@ static int tracker_check_response(ConnectionInfo *pTrackerServer,
char *pFlags;
if ((result=tcprecvdata_nb(pTrackerServer->sock, &resp, \
sizeof(resp), g_fdfs_network_timeout)) != 0)
sizeof(resp), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, recv data fail, " \
@ -1336,7 +1336,7 @@ static int tracker_check_response(ConnectionInfo *pTrackerServer,
}
if ((result=tcprecvdata_nb(pTrackerServer->sock, in_buff, \
nInPackLen, g_fdfs_network_timeout)) != 0)
nInPackLen, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, recv data fail, " \
@ -1582,7 +1582,7 @@ int tracker_sync_src_req(ConnectionInfo *pTrackerServer, \
strcpy(out_buff + sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN, \
pReader->storage_id);
if ((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \
@ -1644,7 +1644,7 @@ static int tracker_sync_dest_req(ConnectionInfo *pTrackerServer)
memset(&header, 0, sizeof(header));
header.cmd = TRACKER_PROTO_CMD_STORAGE_SYNC_DEST_REQ;
if ((result=tcpsenddata_nb(pTrackerServer->sock, &header, \
sizeof(header), g_fdfs_network_timeout)) != 0)
sizeof(header), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \
@ -1701,7 +1701,7 @@ static int tracker_sync_dest_query(ConnectionInfo *pTrackerServer)
memset(&header, 0, sizeof(header));
header.cmd = TRACKER_PROTO_CMD_STORAGE_SYNC_DEST_QUERY;
if ((result=tcpsenddata_nb(pTrackerServer->sock, &header, \
sizeof(header), g_fdfs_network_timeout)) != 0)
sizeof(header), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \
@ -1763,7 +1763,7 @@ static int tracker_report_trunk_fid(ConnectionInfo *pTrackerServer)
int2buff(g_current_trunk_file_id, out_buff + sizeof(TrackerHeader));
if ((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \
@ -1809,7 +1809,7 @@ static int tracker_report_trunk_free_space(ConnectionInfo *pTrackerServer)
long2buff(g_trunk_total_free_space / FDFS_ONE_MB, \
out_buff + sizeof(TrackerHeader));
if ((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \
@ -1855,7 +1855,7 @@ static int tracker_fetch_trunk_fid(ConnectionInfo *pTrackerServer)
memset(out_buff, 0, sizeof(out_buff));
pHeader->cmd = TRACKER_PROTO_CMD_STORAGE_FETCH_TRUNK_FID;
if ((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \
@ -1929,7 +1929,7 @@ static int tracker_sync_notify(ConnectionInfo *pTrackerServer, const int tracker
long2buff(g_sync_until_timestamp, pReqBody->until_timestamp);
if ((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \
@ -2079,7 +2079,7 @@ int tracker_report_join(ConnectionInfo *pTrackerServer, \
long2buff(out_len - (int)sizeof(TrackerHeader), pHeader->pkg_len);
if ((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
out_len, g_fdfs_network_timeout)) != 0)
out_len, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \
@ -2163,7 +2163,7 @@ static int tracker_report_sync_timestamp(ConnectionInfo *pTrackerServer,
}
if((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
sizeof(TrackerHeader) + body_len, g_fdfs_network_timeout)) != 0)
sizeof(TrackerHeader) + body_len, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \
@ -2269,7 +2269,7 @@ static int tracker_report_df_stat(ConnectionInfo *pTrackerServer,
}
result = tcpsenddata_nb(pTrackerServer->sock, pBuff, \
total_len, g_fdfs_network_timeout);
total_len, SF_G_NETWORK_TIMEOUT);
if (pBuff != out_buff)
{
free(pBuff);
@ -2406,7 +2406,7 @@ static int tracker_heart_beat(ConnectionInfo *pTrackerServer,
pHeader->cmd = TRACKER_PROTO_CMD_STORAGE_BEAT;
if((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
sizeof(TrackerHeader) + body_len, g_fdfs_network_timeout)) != 0)
sizeof(TrackerHeader) + body_len, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \
@ -2463,7 +2463,7 @@ static int tracker_storage_change_status(ConnectionInfo *pTrackerServer,
*(out_buff + sizeof(TrackerHeader)) = new_status;
if((result=tcpsenddata_nb(pTrackerServer->sock, out_buff,
sizeof(TrackerHeader) + body_len, g_fdfs_network_timeout)) != 0)
sizeof(TrackerHeader) + body_len, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, "
"tracker server %s:%d, send data fail, "
@ -2510,7 +2510,7 @@ static int tracker_storage_changelog_req(ConnectionInfo *pTrackerServer)
pHeader->cmd = TRACKER_PROTO_CMD_STORAGE_CHANGELOG_REQ;
if((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
sizeof(TrackerHeader), g_fdfs_network_timeout)) != 0)
sizeof(TrackerHeader), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \

4
storage/trunk_mgr/trunk_client.c
View File

@ -51,7 +51,7 @@ static int trunk_client_trunk_do_alloc_space(ConnectionInfo *pTrunkServer, \
pHeader->cmd = STORAGE_PROTO_CMD_TRUNK_ALLOC_SPACE;
if ((result=tcpsenddata_nb(pTrunkServer->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -186,7 +186,7 @@ static int trunk_client_trunk_confirm_or_free(ConnectionInfo *pTrunkServer,\
int2buff(pTrunkInfo->file.size, pTrunkBuff->size);
if ((result=tcpsenddata_nb(pTrunkServer->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \

6
storage/trunk_mgr/trunk_mem.c
View File

@ -301,7 +301,7 @@ int storage_trunk_destroy_ex(const bool bNeedSleep,
"storage trunk destroy", __LINE__);
if (bSaveData)
{
if (g_current_time - g_up_time >= 3600 &&
if (g_current_time - g_sf_global_vars.up_time >= 3600 &&
g_trunk_compress_binlog_interval == 0)
{
result = storage_trunk_save();
@ -958,12 +958,12 @@ static int storage_trunk_compress()
return EAGAIN;
}
if (g_current_time - g_up_time < 600)
if (g_current_time - g_sf_global_vars.up_time < 600)
{
logWarning("file: "__FILE__", line: %d, "
"too little time lapse: %ds afer startup, "
"skip trunk binlog compress", __LINE__,
(int)(g_current_time - g_up_time));
(int)(g_current_time - g_sf_global_vars.up_time));
return EAGAIN;
}

12
storage/trunk_mgr/trunk_sync.c
View File

@ -184,7 +184,7 @@ int trunk_sync_init()
return errno != 0 ? errno : ENOENT;
}
STORAGE_CHOWN(data_path, geteuid(), getegid())
SF_CHOWN_TO_RUNBY_RETURN_ON_ERROR(data_path);
}
snprintf(sync_path, sizeof(sync_path), \
@ -201,7 +201,7 @@ int trunk_sync_init()
return errno != 0 ? errno : ENOENT;
}
STORAGE_CHOWN(sync_path, geteuid(), getegid())
SF_CHOWN_TO_RUNBY_RETURN_ON_ERROR(sync_path);
}
trunk_binlog_write_cache_buff = (char *)malloc( \
@ -227,7 +227,7 @@ int trunk_sync_init()
return result;
}
STORAGE_FCHOWN(trunk_binlog_fd, binlog_filename, geteuid(), getegid())
SF_FCHOWN_TO_RUNBY_RETURN_ON_ERROR(trunk_binlog_fd, binlog_filename);
return 0;
}
@ -1717,7 +1717,7 @@ static int trunk_write_to_mark_file(TrunkBinLogReader *pReader)
if ((result=safeWriteToFile(pReader->mark_filename, buff, len)) == 0)
{
STORAGE_CHOWN(pReader->mark_filename, geteuid(), getegid())
SF_CHOWN_TO_RUNBY_RETURN_ON_ERROR(pReader->mark_filename);
pReader->last_binlog_offset = pReader->binlog_offset;
}
@ -2008,7 +2008,7 @@ static int trunk_sync_data(TrunkBinLogReader *pReader, \
long2buff(length, header.pkg_len);
header.cmd = STORAGE_PROTO_CMD_TRUNK_SYNC_BINLOG;
if ((result=tcpsenddata_nb(pStorage->sock, &header, \
sizeof(TrackerHeader), g_fdfs_network_timeout)) != 0)
sizeof(TrackerHeader), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("FILE: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \
@ -2019,7 +2019,7 @@ static int trunk_sync_data(TrunkBinLogReader *pReader, \
}
if ((result=tcpsenddata_nb(pStorage->sock, pReader->binlog_buff.buffer,\
length, g_fdfs_network_timeout)) != 0)
length, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("FILE: "__FILE__", line: %d, " \
"send data to storage server %s:%d fail, " \

2
tracker/fdfs_server_id_func.c
View File

@ -573,7 +573,7 @@ int fdfs_get_storage_ids_from_tracker_server(TrackerServerInfo *pTrackerServer)
{
int2buff(start_index, p);
if ((result=tcpsenddata_nb(conn->sock, out_buff,
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, "
"send data to tracker server %s:%d fail, "

2
tracker/fdfs_shared_func.c
View File

@ -413,7 +413,7 @@ int fdfs_connection_pool_init(const char *config_filename, \
return EINVAL;
}
return conn_pool_init(&g_connection_pool, g_fdfs_connect_timeout, \
return conn_pool_init(&g_connection_pool, SF_G_CONNECT_TIMEOUT, \
0, g_connection_pool_max_idle_time);
}

2
tracker/fdfs_trackerd.c
View File

@ -434,7 +434,7 @@ static void sigAlarmHandler(int sig)
server.port = SF_G_INNER_PORT;
server.sock = -1;
if (conn_pool_connect_server(&server, g_fdfs_connect_timeout) != 0)
if (conn_pool_connect_server(&server, SF_G_CONNECT_TIMEOUT) != 0)
{
return;
}

8
tracker/tracker_dump.c
View File

@ -291,8 +291,8 @@ static int fdfs_dump_global_vars(char *buff, const int buffSize)
char *p;
total_len = snprintf(buff, buffSize,
"g_fdfs_connect_timeout=%ds\n"
"g_fdfs_network_timeout=%ds\n"
"SF_G_CONNECT_TIMEOUT=%ds\n"
"SF_G_NETWORK_TIMEOUT=%ds\n"
"SF_G_BASE_PATH_STR=%s\n"
"g_fdfs_version=%d.%02d\n"
"SF_G_CONTINUE_FLAG=%d\n"
@ -345,8 +345,8 @@ static int fdfs_dump_global_vars(char *buff, const int buffSize)
#if defined(DEBUG_FLAG) && defined(OS_LINUX)
"g_exe_name=%s\n"
#endif
, g_fdfs_connect_timeout
, g_fdfs_network_timeout
, SF_G_CONNECT_TIMEOUT
, SF_G_NETWORK_TIMEOUT
, SF_G_BASE_PATH_STR
, g_fdfs_version.major, g_fdfs_version.minor
, SF_G_CONTINUE_FLAG

11
tracker/tracker_func.c
View File

@ -171,9 +171,10 @@ int tracker_load_from_conf_file(const char *filename)
sf_set_current_time();
SF_SET_CONTEXT_INI_CONFIG(config, filename, &iniContext,
SF_SET_CONTEXT_INI_CONFIG_EX(config, filename, &iniContext,
NULL, FDFS_TRACKER_SERVER_DEF_PORT,
FDFS_TRACKER_SERVER_DEF_PORT, DEFAULT_WORK_THREADS);
FDFS_TRACKER_SERVER_DEF_PORT, DEFAULT_WORK_THREADS,
"buff_size");
if ((result=sf_load_config_ex("trackerd", &config,
task_buffer_extra_size, need_set_run_by)) != 0)
{
@ -475,7 +476,8 @@ int tracker_load_from_conf_file(const char *filename)
#endif
if (g_if_use_trunk_file && g_groups.store_server == FDFS_STORE_SERVER_ROUND_ROBIN)
if (g_if_use_trunk_file && g_groups.store_server ==
FDFS_STORE_SERVER_ROUND_ROBIN)
{
logInfo("file: "__FILE__", line: %d, "
"set store_server to %d because use_trunk_file is true",
@ -483,7 +485,8 @@ int tracker_load_from_conf_file(const char *filename)
g_groups.store_server = FDFS_STORE_SERVER_FIRST_BY_IP;
}
sf_global_config_to_string(sz_global_config, sizeof(sz_global_config));
sf_global_config_to_string(sz_global_config,
sizeof(sz_global_config));
sf_context_config_to_string(&g_sf_context,
sz_service_config, sizeof(sz_service_config));

6
tracker/tracker_http_check.c
View File

@ -66,7 +66,7 @@ static void *http_check_entrance(void *arg)
{
sock = socketClientAuto((*ppServer)->ip_addr,
(*ppGroup)->storage_http_port,
g_fdfs_connect_timeout, O_NONBLOCK, &result);
SF_G_CONNECT_TIMEOUT, O_NONBLOCK, &result);
if (sock >= 0)
{
close(sock);
@ -135,8 +135,8 @@ static void *http_check_entrance(void *arg)
sprintf(url, "http://%s:%d%s", (*ppServer)->ip_addr, \
(*ppGroup)->storage_http_port, g_http_check_uri);
result = get_url_content(url, g_fdfs_connect_timeout, \
g_fdfs_network_timeout, &http_status, \
result = get_url_content(url, SF_G_CONNECT_TIMEOUT, \
SF_G_NETWORK_TIMEOUT, &http_status, \
&content, &content_len, error_info);
if (g_http_servers_dirty)

4
tracker/tracker_mem.c
View File

@ -3823,7 +3823,7 @@ static int tracker_mem_get_sys_file_piece(ConnectionInfo *pTrackerServer, \
*p++ = file_index;
long2buff(*offset, p);
if ((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"send data to tracker server %s:%d fail, " \
@ -4989,7 +4989,7 @@ static int _storage_get_trunk_binlog_size(
memset(out_buff, 0, sizeof(out_buff));
pHeader->cmd = STORAGE_PROTO_CMD_TRUNK_GET_BINLOG_SIZE;
if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"storage server %s:%d, send data fail, " \

14
tracker/tracker_proto.c
View File

@ -133,7 +133,7 @@ int fdfs_recv_response(ConnectionInfo *pTrackerServer, \
}
if ((result=tcprecvdata_nb(pTrackerServer->sock, *buff, \
*in_bytes, g_fdfs_network_timeout)) != 0)
*in_bytes, SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server: %s:%d, recv data fail, " \
@ -161,7 +161,7 @@ int fdfs_quit(ConnectionInfo *pTrackerServer)
memset(&header, 0, sizeof(header));
header.cmd = FDFS_PROTO_CMD_QUIT;
result = tcpsenddata_nb(pTrackerServer->sock, &header, \
sizeof(header), g_fdfs_network_timeout);
sizeof(header), SF_G_NETWORK_TIMEOUT);
if(result != 0)
{
logError("file: "__FILE__", line: %d, " \
@ -185,7 +185,7 @@ int fdfs_deal_no_body_cmd(ConnectionInfo *pTrackerServer, const int cmd)
memset(&header, 0, sizeof(header));
header.cmd = cmd;
result = tcpsenddata_nb(pTrackerServer->sock, &header, \
sizeof(header), g_fdfs_network_timeout);
sizeof(header), SF_G_NETWORK_TIMEOUT);
if(result != 0)
{
logError("file: "__FILE__", line: %d, " \
@ -499,7 +499,7 @@ ConnectionInfo *tracker_connect_server_no_pool_ex(TrackerServerInfo *pServerInfo
}
*err_no = conn_pool_connect_server_ex(pServerInfo->connections
+ pServerInfo->index, g_fdfs_connect_timeout,
+ pServerInfo->index, SF_G_CONNECT_TIMEOUT,
bind_addr, log_connect_error);
if (*err_no == 0)
{
@ -518,7 +518,7 @@ ConnectionInfo *tracker_connect_server_no_pool_ex(TrackerServerInfo *pServerInfo
if (current_index != pServerInfo->index)
{
if ((*err_no=conn_pool_connect_server_ex(conn,
g_fdfs_connect_timeout, bind_addr,
SF_G_CONNECT_TIMEOUT, bind_addr,
log_connect_error)) == 0)
{
pServerInfo->index = current_index;
@ -600,7 +600,7 @@ static int fdfs_do_parameter_req(ConnectionInfo *pTrackerServer, \
pHeader = (TrackerHeader *)out_buff;
pHeader->cmd = TRACKER_PROTO_CMD_STORAGE_PARAMETER_REQ;
if((result=tcpsenddata_nb(pTrackerServer->sock, out_buff, \
sizeof(TrackerHeader), g_fdfs_network_timeout)) != 0)
sizeof(TrackerHeader), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, " \
"tracker server %s:%d, send data fail, " \
@ -740,7 +740,7 @@ int fdfs_get_tracker_status(TrackerServerInfo *pTrackerServer,
memset(&header, 0, sizeof(header));
header.cmd = TRACKER_PROTO_CMD_TRACKER_GET_STATUS;
if ((result=tcpsenddata_nb(conn->sock, &header,
sizeof(header), g_fdfs_network_timeout)) != 0)
sizeof(header), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, "
"send data to tracker server %s:%d fail, "

6
tracker/tracker_proto.h
View File

@ -217,10 +217,10 @@ extern "C" {
#endif
#define tracker_connect_server(pServerInfo, err_no) \
tracker_connect_server_ex(pServerInfo, g_fdfs_connect_timeout, err_no)
tracker_connect_server_ex(pServerInfo, SF_G_CONNECT_TIMEOUT, err_no)
#define tracker_make_connection(conn, err_no) \
tracker_make_connection_ex(conn, g_fdfs_connect_timeout, err_no)
tracker_make_connection_ex(conn, SF_G_CONNECT_TIMEOUT, err_no)
int fdfs_set_body_length(struct fast_task_info *pTask);
@ -297,7 +297,7 @@ static inline int fdfs_recv_header(ConnectionInfo *pTrackerServer,
int64_t *in_bytes)
{
return fdfs_recv_header_ex(pTrackerServer,
g_fdfs_network_timeout, in_bytes);
SF_G_NETWORK_TIMEOUT, in_bytes);
}
int fdfs_recv_response(ConnectionInfo *pTrackerServer, \

4
tracker/tracker_relationship.c
View File

@ -50,7 +50,7 @@ static int fdfs_ping_leader(ConnectionInfo *pTrackerServer)
memset(&header, 0, sizeof(header));
header.cmd = TRACKER_PROTO_CMD_TRACKER_PING_LEADER;
result = tcpsenddata_nb(pTrackerServer->sock, &header, \
sizeof(header), g_fdfs_network_timeout);
sizeof(header), SF_G_NETWORK_TIMEOUT);
if(result != 0)
{
logError("file: "__FILE__", line: %d, "
@ -280,7 +280,7 @@ static int do_notify_leader_changed(TrackerServerInfo *pTrackerServer, \
pLeader->ip_addr, pLeader->port);
long2buff(FDFS_PROTO_IP_PORT_SIZE, pHeader->pkg_len);
if ((result=tcpsenddata_nb(conn->sock, out_buff, \
sizeof(out_buff), g_fdfs_network_timeout)) != 0)
sizeof(out_buff), SF_G_NETWORK_TIMEOUT)) != 0)
{
logError("file: "__FILE__", line: %d, "
"send data to tracker server %s:%d fail, "

30
tracker/tracker_service.c
View File

@ -45,6 +45,8 @@ static int lock_by_client_count = 0;
static void tracker_find_max_free_space_group();
static int tracker_deal_task(struct fast_task_info *pTask, const int stage);
static void task_finish_clean_up(struct fast_task_info *pTask)
{
TrackerClientInfo *pClientInfo;
@ -63,6 +65,25 @@ static void task_finish_clean_up(struct fast_task_info *pTask)
sf_task_finish_clean_up(pTask);
}
static int sock_accept_done_callback(struct fast_task_info *task,
const in_addr_t client_addr, const bool bInnerPort)
{
if (g_allow_ip_count >= 0)
{
if (bsearch(&client_addr, g_allow_ip_addrs,
g_allow_ip_count, sizeof(in_addr_t),
cmp_by_ip_addr_t) == NULL)
{
logError("file: "__FILE__", line: %d, "
"ip addr %s is not allowed to access",
__LINE__, task->client_ip);
return EPERM;
}
}
return 0;
}
int tracker_service_init()
{
int result;
@ -72,9 +93,10 @@ int tracker_service_init()
return result;
}
result = sf_service_init("fdfs_trackerd", NULL, NULL, NULL,
fdfs_set_body_length, tracker_deal_task, task_finish_clean_up,
NULL, 1000, sizeof(TrackerHeader), sizeof(TrackerClientInfo));
result = sf_service_init("fdfs_trackerd", NULL, NULL,
sock_accept_done_callback, fdfs_set_body_length,
tracker_deal_task, task_finish_clean_up, NULL, 1000,
sizeof(TrackerHeader), sizeof(TrackerClientInfo));
sf_enable_thread_notify(false);
sf_set_remove_from_ready_list(false);
return result;
@ -3651,7 +3673,7 @@ static int tracker_deal_storage_beat(struct fast_task_info *pTask)
} \
int tracker_deal_task(struct fast_task_info *pTask, const int stage)
static int tracker_deal_task(struct fast_task_info *pTask, const int stage)
{
TrackerHeader *pHeader;
int result;

3
tracker/tracker_service.h
View File

@ -25,9 +25,6 @@ extern "C" {
int tracker_service_init();
void tracker_service_destroy();
void tracker_accept_loop(int server_sock);
int tracker_deal_task(struct fast_task_info *pTask, const int stage);
#ifdef __cplusplus
}
#endif