From: <n-...@us...> - 2011-08-04 02:00:55
|
Revision: 5447 http://gfarm.svn.sourceforge.net/gfarm/?rev=5447&view=rev Author: n-soda Date: 2011-08-04 02:00:48 +0000 (Thu, 04 Aug 2011) Log Message: ----------- when a slave gfmd is going to be upgraded to a master gfmd, make sure that all journal records are applied to the backend database, before calling dead_file_copy_init_load(). Modified Paths: -------------- gfarm_v2/trunk/server/gfmd/db_journal.c gfarm_v2/trunk/server/gfmd/db_journal.h gfarm_v2/trunk/server/gfmd/gfmd.c gfarm_v2/trunk/server/gfmd/journal_file.c gfarm_v2/trunk/server/gfmd/journal_file.h Modified: gfarm_v2/trunk/server/gfmd/db_journal.c =================================================================== --- gfarm_v2/trunk/server/gfmd/db_journal.c 2011-08-04 00:55:40 UTC (rev 5446) +++ gfarm_v2/trunk/server/gfmd/db_journal.c 2011-08-04 02:00:48 UTC (rev 5447) @@ -3404,9 +3404,9 @@ } void -db_journal_cancel_read(void) +db_journal_wait_for_apply_thread(void) { - journal_file_cancel_read(journal_file_main_reader(self_jf)); + journal_file_wait_for_read_completion(journal_file_main_reader(self_jf)); } static gfarm_error_t Modified: gfarm_v2/trunk/server/gfmd/db_journal.h =================================================================== --- gfarm_v2/trunk/server/gfmd/db_journal.h 2011-08-04 00:55:40 UTC (rev 5446) +++ gfarm_v2/trunk/server/gfmd/db_journal.h 2011-08-04 02:00:48 UTC (rev 5447) @@ -14,7 +14,7 @@ gfarm_error_t db_journal_read(struct journal_file_reader *, void *, gfarm_error_t (*)(void *, gfarm_uint64_t, enum journal_operation, void *, void *, size_t, int *), void *, int *); -void db_journal_cancel_read(void); +void db_journal_wait_for_apply_thread(void); gfarm_error_t db_journal_reader_reopen(struct journal_file_reader **, gfarm_uint64_t); gfarm_error_t db_journal_fetch(struct journal_file_reader *, gfarm_uint64_t, Modified: gfarm_v2/trunk/server/gfmd/gfmd.c =================================================================== --- gfarm_v2/trunk/server/gfmd/gfmd.c 2011-08-04 00:55:40 UTC (rev 5446) +++ gfarm_v2/trunk/server/gfmd/gfmd.c 2011-08-04 02:00:48 UTC (rev 5447) @@ -1067,18 +1067,26 @@ mdhost_disconnect(master, NULL); gflog_info(GFARM_MSG_UNFIXED, "start transforming to the master gfmd ..."); + + db_journal_cancel_recvq(); + + /* + * wait for data transfer from the journal to the backend DB. + * this must be done before dead_file_copy_init_load(). + */ + db_journal_wait_for_apply_thread(); + giant_lock(); mdhost_set_self_as_master(); + + /* this must be after db_journal_wait_for_apply_thread() */ dead_file_copy_init_load(); giant_unlock(); gfarm_cond_signal(&transform_cond, diag, TRANSFORM_COND_DIAG); - db_journal_cancel_read(); - db_journal_cancel_recvq(); - start_db_journal_threads(); start_gfmdc_threads(); Modified: gfarm_v2/trunk/server/gfmd/journal_file.c =================================================================== --- gfarm_v2/trunk/server/gfmd/journal_file.c 2011-08-04 00:55:40 UTC (rev 5446) +++ gfarm_v2/trunk/server/gfmd/journal_file.c 2011-08-04 02:00:48 UTC (rev 5447) @@ -62,15 +62,15 @@ #define JOURNAL_FILE_READER_F_BLOCK_WRITER 0x1 #define JOURNAL_FILE_READER_F_WRAP 0x2 #define JOURNAL_FILE_READER_F_INVALID 0x4 -#define JOURNAL_FILE_READER_F_CANCEL 0x8 +#define JOURNAL_FILE_READER_F_DRAIN 0x8 #define JOURNAL_FILE_READER_IS_BLOCK_WRITER(r) \ (((r)->flags & JOURNAL_FILE_READER_F_BLOCK_WRITER) != 0) #define JOURNAL_FILE_READER_IS_WRAP(r) \ (((r)->flags & JOURNAL_FILE_READER_F_WRAP) != 0) #define JOURNAL_FILE_READER_IS_INVALID(r) \ (((r)->flags & JOURNAL_FILE_READER_F_INVALID) != 0) -#define JOURNAL_FILE_READER_CANCELED(r) \ - (((r)->flags & JOURNAL_FILE_READER_F_CANCEL) != 0) +#define JOURNAL_FILE_READER_DRAINED(r) \ + (((r)->flags & JOURNAL_FILE_READER_F_DRAIN) != 0) #define JOURNAL_RECORD_SIZE_MAX (1024 * 1024) #define JOURNAL_FILE_HEADER_SIZE 4096 @@ -93,7 +93,7 @@ size_t size, max_size; off_t tail; int wait_until_nonempty; - pthread_cond_t nonfull_cond, nonempty_cond, cancel_cond; + pthread_cond_t nonfull_cond, nonempty_cond, drain_cond; pthread_mutex_t mutex, pos_mutex; }; @@ -1191,7 +1191,7 @@ gfarm_cond_init(&jf->nonfull_cond, diag, JOURNAL_FILE_STR); gfarm_cond_init(&jf->nonempty_cond, diag, JOURNAL_FILE_STR); - gfarm_cond_init(&jf->cancel_cond, diag, JOURNAL_FILE_STR); + gfarm_cond_init(&jf->drain_cond, diag, JOURNAL_FILE_STR); gfarm_mutex_init(&jf->mutex, diag, JOURNAL_FILE_STR); gfarm_mutex_init(&jf->pos_mutex, diag, JOURNAL_FILE_STR); jf->wait_until_nonempty = 0; @@ -1214,7 +1214,7 @@ void journal_file_reader_close(struct journal_file_reader *reader) { - journal_file_reader_set_flag(reader, JOURNAL_FILE_READER_F_CANCEL, 1); + journal_file_reader_set_flag(reader, JOURNAL_FILE_READER_F_DRAIN, 1); if (reader->xdr) { gfp_xdr_free(reader->xdr); reader->xdr = NULL; @@ -1528,7 +1528,7 @@ gfarm_error_t e; gfarm_uint64_t seqnum; enum journal_operation ope = 0; - int needs_free = 1, canceled = 0; + int needs_free = 1, drained = 0; size_t len; void *obj = NULL; struct journal_file *jf = reader->file; @@ -1563,16 +1563,16 @@ } while (avail < min_rec_size) { jf->wait_until_nonempty = 1; - gfarm_cond_wait(&jf->nonempty_cond, &jf->mutex, - diag, JOURNAL_FILE_STR); - jf->wait_until_nonempty = 0; - if (JOURNAL_FILE_READER_CANCELED(reader)) { + if (JOURNAL_FILE_READER_DRAINED(reader)) { journal_file_reader_set_flag(reader, - JOURNAL_FILE_READER_F_CANCEL, 0); - canceled = 1; + JOURNAL_FILE_READER_F_DRAIN, 0); + drained = 1; e = GFARM_ERR_CANT_OPEN; goto unlock; } + gfarm_cond_wait(&jf->nonempty_cond, &jf->mutex, + diag, JOURNAL_FILE_STR); + jf->wait_until_nonempty = 0; if ((e = gfp_xdr_recv_ahead(xdr, JOURNAL_READ_AHEAD_SIZE, &avail)) != GFARM_ERR_NO_ERROR) { @@ -1609,8 +1609,8 @@ journal_file_mutex_unlock(jf, diag); if (needs_free && obj) free_op(op_arg, ope, obj); - if (canceled) - gfarm_cond_signal(&jf->cancel_cond, diag, JOURNAL_FILE_STR); + if (drained) + gfarm_cond_signal(&jf->drain_cond, diag, JOURNAL_FILE_STR); return (e); } @@ -1700,19 +1700,19 @@ } void -journal_file_cancel_read(struct journal_file_reader *reader) +journal_file_wait_for_read_completion(struct journal_file_reader *reader) { struct journal_file *jf = reader->file; - static const char *diag = "journal_file_cancel_read"; + static const char *diag = "journal_file_wait_for_read_completion"; journal_file_mutex_lock(jf, diag); - journal_file_reader_set_flag(reader, JOURNAL_FILE_READER_F_CANCEL, 1); + journal_file_reader_set_flag(reader, JOURNAL_FILE_READER_F_DRAIN, 1); journal_file_mutex_unlock(jf, diag); gfarm_cond_signal(&jf->nonempty_cond, diag, JOURNAL_FILE_STR); journal_file_mutex_lock(jf, diag); - while (JOURNAL_FILE_READER_CANCELED(reader)) - gfarm_cond_wait(&jf->cancel_cond, &jf->mutex, diag, + while (JOURNAL_FILE_READER_DRAINED(reader)) + gfarm_cond_wait(&jf->drain_cond, &jf->mutex, diag, JOURNAL_FILE_STR); journal_file_mutex_unlock(jf, diag); } Modified: gfarm_v2/trunk/server/gfmd/journal_file.h =================================================================== --- gfarm_v2/trunk/server/gfmd/journal_file.h 2011-08-04 00:55:40 UTC (rev 5446) +++ gfarm_v2/trunk/server/gfmd/journal_file.h 2011-08-04 02:00:48 UTC (rev 5447) @@ -102,7 +102,7 @@ void *, int *); gfarm_error_t journal_file_read_serialized(struct journal_file_reader *, char **, gfarm_uint32_t *, gfarm_uint64_t *, int *); -void journal_file_cancel_read(struct journal_file_reader *); +void journal_file_wait_for_read_completion(struct journal_file_reader *); void journal_file_wait_until_empty(struct journal_file *); int journal_file_is_closed(struct journal_file *); int journal_file_is_waiting_until_nonempty(struct journal_file *); This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site. |