diff options
Diffstat (limited to 'xlators/features/changelog/src/changelog-helpers.c')
-rw-r--r-- | xlators/features/changelog/src/changelog-helpers.c | 211 |
1 files changed, 117 insertions, 94 deletions
diff --git a/xlators/features/changelog/src/changelog-helpers.c b/xlators/features/changelog/src/changelog-helpers.c index 91c43a16c..ad4fe4013 100644 --- a/xlators/features/changelog/src/changelog-helpers.c +++ b/xlators/features/changelog/src/changelog-helpers.c @@ -21,7 +21,6 @@ #include "changelog-helpers.h" #include "changelog-mem-types.h" -#include "changelog-encoders.h" #include <pthread.h> void @@ -53,51 +52,45 @@ changelog_thread_cleanup (xlator_t *this, pthread_t thr_id) inline void * changelog_get_usable_buffer (changelog_local_t *local) { - changelog_log_data_t *cld = NULL; + changelog_write_data_t *cwd = &local->cld.cld_wdata; - if (!local) - return NULL; - - cld = &local->cld; - if (!cld->cld_iobuf) + if (!cwd->cwd_iobuf) return NULL; - return cld->cld_iobuf->ptr; + return cwd->cwd_ptr; } inline void changelog_set_usable_record_and_length (changelog_local_t *local, size_t len, int xr) { - changelog_log_data_t *cld = NULL; + changelog_write_data_t *cwd = &local->cld.cld_wdata; - cld = &local->cld; - - cld->cld_ptr_len = len; - cld->cld_xtra_records = xr; + cwd->cwd_ptr_len = len; + cwd->cwd_xtra_records = xr; } void changelog_local_cleanup (xlator_t *xl, changelog_local_t *local) { - int i = 0; - changelog_opt_t *co = NULL; - changelog_log_data_t *cld = NULL; + int i = 0; + changelog_opt_t *co = NULL; + changelog_write_data_t *cwd = NULL; if (!local) return; - cld = &local->cld; + cwd = &local->cld.cld_wdata; /* cleanup dynamic allocation for extra records */ - if (cld->cld_xtra_records) { - co = (changelog_opt_t *) cld->cld_ptr; - for (; i < cld->cld_xtra_records; i++, co++) + if (cwd->cwd_xtra_records) { + co = (changelog_opt_t *) cwd->cwd_ptr; + for (; i < cwd->cwd_xtra_records; i++, co++) if (co->co_free) co->co_free (co); } - CHANGELOG_IOBUF_UNREF (cld->cld_iobuf); + CHANGELOG_IOBUF_UNREF (cwd->cwd_iobuf); if (local->inode) inode_unref (local->inode); @@ -125,7 +118,8 @@ changelog_write (int fd, char *buffer, size_t len) static int changelog_rollover_changelog (xlator_t *this, - changelog_priv_t *priv, unsigned long ts) + changelog_priv_t *priv, + changelog_rollover_data_t *crd) { int ret = -1; int notify = 0; @@ -138,11 +132,22 @@ changelog_rollover_changelog (xlator_t *this, priv->changelog_fd = -1; } + /** + * no rolling-over of changelogs, policy implementer choose + * to do the heavy-lifting of having distinct changelog name. + * + * NOTE: This implies libgfchangelog would not be notified + (well, we could, but lets not do that now...) + */ + if (!crd->crd_use_suffix) + return 0; + (void) snprintf (ofile, PATH_MAX, - "%s/"CHANGELOG_FILE_NAME, priv->changelog_dir); - (void) snprintf (nfile, PATH_MAX, - "%s/"CHANGELOG_FILE_NAME".%lu", - priv->changelog_dir, ts); + "%s/%s", priv->changelog_dir, + crd->crd_changelog_oname); + (void) snprintf (nfile, PATH_MAX, "%s/%s.%lu", + priv->changelog_dir, + crd->crd_changelog_name, crd->crd_roll_key); ret = rename (ofile, nfile); if (!ret) @@ -174,7 +179,8 @@ changelog_rollover_changelog (xlator_t *this, int changelog_open (xlator_t *this, - changelog_priv_t *priv) + changelog_priv_t *priv, + changelog_local_t *local, changelog_rollover_data_t *crd) { int fd = 0; int ret = -1; @@ -183,12 +189,12 @@ changelog_open (xlator_t *this, char changelog_path[PATH_MAX] = {0,}; (void) snprintf (changelog_path, PATH_MAX, - "%s/"CHANGELOG_FILE_NAME, - priv->changelog_dir); + "%s/%s", priv->changelog_dir, + crd->crd_changelog_name); flags |= (O_CREAT | O_RDWR); if (priv->fsync_interval == 0) - flags |= O_SYNC; + flags |= O_SYNC; fd = open (changelog_path, flags, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH); @@ -201,12 +207,25 @@ changelog_open (xlator_t *this, } priv->changelog_fd = fd; + CHANGELOG_INVOKE_CFOP (this, priv, reset_offset, local); + + /* preallocate if required */ + if (crd->crd_prealloc_size > 0) { + ret = posix_fallocate (priv->changelog_fd, + 0, crd->crd_prealloc_size); + if (ret) { + gf_log (this->name, GF_LOG_ERROR, + "failed to preallocate %llu bytes", + (unsigned long long) crd->crd_prealloc_size); + } + } (void) snprintf (buffer, 1024, CHANGELOG_HEADER, CHANGELOG_VERSION_MAJOR, CHANGELOG_VERSION_MINOR, - priv->ce->encoder); - ret = changelog_write_change (priv, buffer, strlen (buffer)); + priv->encode_mode); + ret = changelog_write_change (this, priv, + local, buffer, strlen (buffer)); if (ret) { close (priv->changelog_fd); priv->changelog_fd = -1; @@ -219,18 +238,19 @@ changelog_open (xlator_t *this, return ret; } -int +static int changelog_start_next_change (xlator_t *this, changelog_priv_t *priv, - unsigned long ts, gf_boolean_t finale) + changelog_local_t *local, + changelog_log_data_t *cld) { - int ret = -1; - - ret = changelog_rollover_changelog (this, priv, ts); + int ret = 0; + changelog_rollover_data_t *crd = &cld->cld_roll; - if (!ret && !finale) - ret = changelog_open (this, priv); + ret = changelog_rollover_changelog (this, priv, crd); + if (!ret && !crd->crd_finale) + ret = changelog_open (this, priv, local, crd); return ret; } @@ -244,37 +264,42 @@ changelog_entry_length () } int -changelog_fill_rollover_data (changelog_log_data_t *cld, gf_boolean_t is_last) +changelog_write_change (xlator_t *this, changelog_priv_t *priv, + changelog_local_t *local, char *buffer, size_t len) { - struct timeval tv = {0,}; + int ret = -1; + off_t offset = 0; + ssize_t size = 0; + size_t writen = 0; - cld->cld_type = CHANGELOG_TYPE_ROLLOVER; + offset = CHANGELOG_INVOKE_CFOP (this, priv, get_offset, local); - if (gettimeofday (&tv, NULL)) - return -1; + while (writen < len) { + size = pwrite (priv->changelog_fd, + buffer + writen, len - writen, offset + writen); + if (size <= 0) + break; - cld->cld_roll_time = (unsigned long) tv.tv_sec; - cld->cld_finale = is_last; - return 0; -} + writen += size; + } -int -changelog_write_change (changelog_priv_t *priv, char *buffer, size_t len) -{ - return changelog_write (priv->changelog_fd, buffer, len); + if (writen == len) { + ret = 0; + CHANGELOG_INVOKE_CFOP (this, priv, set_offset, local, writen); + } + + return ret; } inline int changelog_handle_change (xlator_t *this, - changelog_priv_t *priv, changelog_log_data_t *cld) + changelog_priv_t *priv, + changelog_local_t *local, changelog_log_data_t *cld) { int ret = 0; if (CHANGELOG_TYPE_IS_ROLLOVER (cld->cld_type)) { - changelog_encode_change(priv); - ret = changelog_start_next_change (this, priv, - cld->cld_roll_time, - cld->cld_finale); + ret = changelog_start_next_change (this, priv, local, cld); if (ret) gf_log (this->name, GF_LOG_ERROR, "Problem rolling over changelog(s)"); @@ -298,7 +323,7 @@ changelog_handle_change (xlator_t *this, goto out; } - ret = priv->ce->encode (this, cld); + ret = priv->ce->encode (this, local, cld); if (ret) { gf_log (this->name, GF_LOG_ERROR, "error writing changelog to disk"); @@ -308,6 +333,17 @@ changelog_handle_change (xlator_t *this, return ret; } +static inline void +changelog_local_init_defaults (changelog_local_t *local, + uuid_t gfid, struct iobuf *iobuf) +{ + changelog_write_data_t *cwd = &(local->cld.cld_wdata); + + uuid_copy (cwd->cwd_gfid, gfid); + cwd->cwd_iobuf = iobuf; + cwd->cwd_xtra_records = 0; /* set by the caller */ +} + changelog_local_t * changelog_local_init (xlator_t *this, inode_t *inode, uuid_t gfid, int xtra_records, @@ -317,7 +353,7 @@ changelog_local_init (xlator_t *this, inode_t *inode, struct iobuf *iobuf = NULL; /** - * We relax the presence of inode if @update_flag is true. + * Relax the presence of inode if @update_flag is true. * The caller (implmentation of the fop) needs to be careful to * not blindly use local->inode. */ @@ -342,10 +378,7 @@ changelog_local_init (xlator_t *this, inode_t *inode, local->update_no_check = update_flag; - uuid_copy (local->cld.cld_gfid, gfid); - - local->cld.cld_iobuf = iobuf; - local->cld.cld_xtra_records = 0; /* set by the caller */ + (void) changelog_local_init_defaults (local, gfid, iobuf); if (inode) local->inode = inode_ref (inode); @@ -373,9 +406,11 @@ changelog_forget (xlator_t *this, inode_t *inode) int changelog_inject_single_event (xlator_t *this, changelog_priv_t *priv, + changelog_local_t *local, changelog_log_data_t *cld) { - return priv->cd.dispatchfn (this, priv, priv->cd.cd_data, cld, NULL); + return priv->cd.dispatchfn (this, priv, + priv->cd.cd_data, local, cld); } /** @@ -386,9 +421,9 @@ void * changelog_rollover (void *data) { int ret = 0; + char *cname = NULL; xlator_t *this = NULL; struct timeval tv = {0,}; - changelog_log_data_t cld = {0,}; changelog_time_slice_t *slice = NULL; changelog_priv_t *priv = data; @@ -403,16 +438,11 @@ changelog_rollover (void *data) if (ret) continue; - ret = changelog_fill_rollover_data (&cld, _gf_false); - if (ret) { - gf_log (this->name, GF_LOG_ERROR, - "failed to fill rollover data"); - continue; - } - LOCK (&priv->lock); { - ret = changelog_inject_single_event (this, priv, &cld); + cname = CHANGELOG_FNAME_FROM_POLICY (priv->cp); + ret = CHANGELOG_INVOKE_CFOP (this, priv, rollover, + cname, _gf_false); if (!ret) SLICE_VERSION_UPDATE (slice); } @@ -428,11 +458,9 @@ changelog_fsync_thread (void *data) int ret = 0; xlator_t *this = NULL; struct timeval tv = {0,}; - changelog_log_data_t cld = {0,}; changelog_priv_t *priv = data; this = priv->cf.this; - cld.cld_type = CHANGELOG_TYPE_FSYNC; while (1) { tv.tv_sec = priv->fsync_interval; @@ -442,7 +470,7 @@ changelog_fsync_thread (void *data) if (ret) continue; - ret = changelog_inject_single_event (this, priv, &cld); + ret = CHANGELOG_INVOKE_CFOP (this, priv, sync); if (ret) gf_log (this->name, GF_LOG_ERROR, "failed to inject fsync event"); @@ -640,19 +668,19 @@ changelog_inode_ctx_get (xlator_t *this, * signifies an update was recorded in the current time slice). */ inline void -changelog_update (xlator_t *this, changelog_priv_t *priv, - changelog_local_t *local, changelog_log_type type) +changelog_update (xlator_t *this, + changelog_priv_t *priv, + changelog_local_t *local, + changelog_log_type type) { - int ret = 0; - unsigned long *iver = NULL; - unsigned long version = 0; - inode_t *inode = NULL; - changelog_time_slice_t *slice = NULL; - changelog_inode_ctx_t *ctx = NULL; - changelog_log_data_t *cld_0 = NULL; - changelog_log_data_t *cld_1 = NULL; - changelog_local_t *next_local = NULL; - gf_boolean_t need_upd = _gf_true; + int ret = 0; + unsigned long *iver = NULL; + unsigned long version = 0; + inode_t *inode = NULL; + changelog_time_slice_t *slice = NULL; + changelog_inode_ctx_t *ctx = NULL; + changelog_log_data_t *cld_0 = NULL; + gf_boolean_t need_upd = _gf_true; slice = &priv->slice; @@ -676,13 +704,8 @@ changelog_update (xlator_t *this, changelog_priv_t *priv, cld_0 = &local->cld; cld_0->cld_type = type; - if ( (next_local = local->prev_entry) != NULL ) { - cld_1 = &next_local->cld; - cld_1->cld_type = type; - } - ret = priv->cd.dispatchfn (this, priv, - priv->cd.cd_data, cld_0, cld_1); + priv->cd.cd_data, local, cld_0); /** * update after the dispatcher has successfully done |