summaryrefslogtreecommitdiffstats
path: root/xlators/performance/io-threads/src/io-threads.c
diff options
context:
space:
mode:
Diffstat (limited to 'xlators/performance/io-threads/src/io-threads.c')
-rw-r--r--xlators/performance/io-threads/src/io-threads.c1968
1 files changed, 1206 insertions, 762 deletions
diff --git a/xlators/performance/io-threads/src/io-threads.c b/xlators/performance/io-threads/src/io-threads.c
index 8201a95a4d0..3d24cc97f4b 100644
--- a/xlators/performance/io-threads/src/io-threads.c
+++ b/xlators/performance/io-threads/src/io-threads.c
@@ -8,281 +8,300 @@
cases as published by the Free Software Foundation.
*/
-#ifndef _CONFIG_H
-#define _CONFIG_H
-#include "config.h"
-#endif
-
-#include "call-stub.h"
-#include "defaults.h"
-#include "glusterfs.h"
-#include "logging.h"
-#include "dict.h"
-#include "xlator.h"
+#include <glusterfs/call-stub.h>
+#include <glusterfs/defaults.h>
+#include <glusterfs/glusterfs.h>
+#include <glusterfs/logging.h>
+#include <glusterfs/dict.h>
+#include <glusterfs/xlator.h>
#include "io-threads.h"
+#include <signal.h>
#include <stdlib.h>
#include <sys/time.h>
#include <time.h>
-#include "locking.h"
+#include <glusterfs/locking.h>
+#include "io-threads-messages.h"
+#include <glusterfs/timespec.h>
-void *iot_worker (void *arg);
-int iot_workers_scale (iot_conf_t *conf);
-int __iot_workers_scale (iot_conf_t *conf);
+void *
+iot_worker(void *arg);
+int
+iot_workers_scale(iot_conf_t *conf);
+int
+__iot_workers_scale(iot_conf_t *conf);
struct volume_options options[];
-#define IOT_FOP(name, frame, this, args ...) \
- do { \
- call_stub_t *__stub = NULL; \
- int __ret = -1; \
+#define IOT_FOP(name, frame, this, args...) \
+ do { \
+ call_stub_t *__stub = NULL; \
+ int __ret = -1; \
\
- __stub = fop_##name##_stub(frame, default_##name##_resume, args); \
- if (!__stub) { \
- __ret = -ENOMEM; \
- goto out; \
- } \
+ __stub = fop_##name##_stub(frame, default_##name##_resume, args); \
+ if (!__stub) { \
+ __ret = -ENOMEM; \
+ goto out; \
+ } \
\
- __ret = iot_schedule (frame, this, __stub); \
+ __ret = iot_schedule(frame, this, __stub); \
\
- out: \
- if (__ret < 0) { \
- default_##name##_failure_cbk (frame, -__ret); \
- if (__stub != NULL) { \
- call_stub_destroy (__stub); \
- } \
- } \
- } while (0)
+ out: \
+ if (__ret < 0) { \
+ default_##name##_failure_cbk(frame, -__ret); \
+ if (__stub != NULL) { \
+ call_stub_destroy(__stub); \
+ } \
+ } \
+ } while (0)
+
+iot_client_ctx_t *
+iot_get_ctx(xlator_t *this, client_t *client)
+{
+ iot_client_ctx_t *ctx = NULL;
+ iot_client_ctx_t *setted_ctx = NULL;
+ int i;
+
+ if (client_ctx_get(client, this, (void **)&ctx) != 0) {
+ ctx = GF_MALLOC(GF_FOP_PRI_MAX * sizeof(*ctx), gf_iot_mt_client_ctx_t);
+ if (ctx) {
+ for (i = 0; i < GF_FOP_PRI_MAX; ++i) {
+ INIT_LIST_HEAD(&ctx[i].clients);
+ INIT_LIST_HEAD(&ctx[i].reqs);
+ }
+ setted_ctx = client_ctx_set(client, this, ctx);
+ if (ctx != setted_ctx) {
+ GF_FREE(ctx);
+ ctx = setted_ctx;
+ }
+ }
+ }
+
+ return ctx;
+}
call_stub_t *
-__iot_dequeue (iot_conf_t *conf, int *pri, struct timespec *sleep)
-{
- call_stub_t *stub = NULL;
- int i = 0;
- struct timeval curtv = {0,}, difftv = {0,};
-
- *pri = -1;
- sleep->tv_sec = 0;
- sleep->tv_nsec = 0;
- for (i = 0; i < IOT_PRI_MAX; i++) {
- if (list_empty (&conf->reqs[i]) ||
- (conf->ac_iot_count[i] >= conf->ac_iot_limit[i]))
- continue;
-
- if (i == IOT_PRI_LEAST) {
- pthread_mutex_lock(&conf->throttle.lock);
- if (!conf->throttle.sample_time.tv_sec) {
- /* initialize */
- gettimeofday(&conf->throttle.sample_time, NULL);
- } else {
- /*
- * Maintain a running count of least priority
- * operations that are handled over a particular
- * time interval. The count is provided via
- * state dump and is used as a measure against
- * least priority op throttling.
- */
- gettimeofday(&curtv, NULL);
- timersub(&curtv, &conf->throttle.sample_time,
- &difftv);
- if (difftv.tv_sec >= IOT_LEAST_THROTTLE_DELAY) {
- conf->throttle.cached_rate =
- conf->throttle.sample_cnt;
- conf->throttle.sample_cnt = 0;
- conf->throttle.sample_time = curtv;
- }
-
- /*
- * If we're over the configured rate limit,
- * provide an absolute time to the caller that
- * represents the soonest we're allowed to
- * return another least priority request.
- */
- if (conf->throttle.rate_limit &&
- conf->throttle.sample_cnt >=
- conf->throttle.rate_limit) {
- struct timeval delay;
- delay.tv_sec = IOT_LEAST_THROTTLE_DELAY;
- delay.tv_usec = 0;
-
- timeradd(&conf->throttle.sample_time,
- &delay, &curtv);
- TIMEVAL_TO_TIMESPEC(&curtv, sleep);
-
- pthread_mutex_unlock(
- &conf->throttle.lock);
- break;
- }
- }
- conf->throttle.sample_cnt++;
- pthread_mutex_unlock(&conf->throttle.lock);
- }
-
- stub = list_entry (conf->reqs[i].next, call_stub_t, list);
- conf->ac_iot_count[i]++;
- *pri = i;
- break;
+__iot_dequeue(iot_conf_t *conf, int *pri)
+{
+ call_stub_t *stub = NULL;
+ int i = 0;
+ iot_client_ctx_t *ctx;
+
+ *pri = -1;
+ for (i = 0; i < GF_FOP_PRI_MAX; i++) {
+ if (conf->ac_iot_count[i] >= conf->ac_iot_limit[i]) {
+ continue;
}
- if (!stub)
- return NULL;
+ if (list_empty(&conf->clients[i])) {
+ continue;
+ }
- conf->queue_size--;
- conf->queue_sizes[*pri]--;
- list_del_init (&stub->list);
+ /* Get the first per-client queue for this priority. */
+ ctx = list_first_entry(&conf->clients[i], iot_client_ctx_t, clients);
+ if (!ctx) {
+ continue;
+ }
- return stub;
-}
+ if (list_empty(&ctx->reqs)) {
+ continue;
+ }
+ /* Get the first request on that queue. */
+ stub = list_first_entry(&ctx->reqs, call_stub_t, list);
+ list_del_init(&stub->list);
+ if (list_empty(&ctx->reqs)) {
+ list_del_init(&ctx->clients);
+ } else {
+ list_rotate_left(&conf->clients[i]);
+ }
-void
-__iot_enqueue (iot_conf_t *conf, call_stub_t *stub, int pri)
-{
- if (pri < 0 || pri >= IOT_PRI_MAX)
- pri = IOT_PRI_MAX-1;
+ conf->ac_iot_count[i]++;
+ conf->queue_marked[i] = _gf_false;
+ *pri = i;
+ break;
+ }
- list_add_tail (&stub->list, &conf->reqs[pri]);
+ if (!stub)
+ return NULL;
- conf->queue_size++;
- conf->queue_sizes[pri]++;
+ conf->queue_size--;
+ conf->queue_sizes[*pri]--;
- return;
+ return stub;
}
+void
+__iot_enqueue(iot_conf_t *conf, call_stub_t *stub, int pri)
+{
+ client_t *client = stub->frame->root->client;
+ iot_client_ctx_t *ctx;
+
+ if (pri < 0 || pri >= GF_FOP_PRI_MAX)
+ pri = GF_FOP_PRI_MAX - 1;
+
+ if (client) {
+ ctx = iot_get_ctx(THIS, client);
+ if (ctx) {
+ ctx = &ctx[pri];
+ }
+ } else {
+ ctx = NULL;
+ }
+ if (!ctx) {
+ ctx = &conf->no_client[pri];
+ }
+
+ if (list_empty(&ctx->reqs)) {
+ list_add_tail(&ctx->clients, &conf->clients[pri]);
+ }
+ list_add_tail(&stub->list, &ctx->reqs);
+
+ conf->queue_size++;
+ GF_ATOMIC_INC(conf->stub_cnt);
+ conf->queue_sizes[pri]++;
+}
void *
-iot_worker (void *data)
-{
- iot_conf_t *conf = NULL;
- xlator_t *this = NULL;
- call_stub_t *stub = NULL;
- struct timespec sleep_till = {0, };
- int ret = 0;
- int pri = -1;
- char timeout = 0;
- char bye = 0;
- struct timespec sleep = {0,};
-
- conf = data;
- this = conf->this;
- THIS = this;
-
- for (;;) {
- sleep_till.tv_sec = time (NULL) + conf->idle_time;
-
- pthread_mutex_lock (&conf->mutex);
- {
- if (pri != -1) {
- conf->ac_iot_count[pri]--;
- pri = -1;
- }
- while (conf->queue_size == 0) {
- conf->sleep_count++;
-
- ret = pthread_cond_timedwait (&conf->cond,
- &conf->mutex,
- &sleep_till);
- conf->sleep_count--;
-
- if (ret == ETIMEDOUT) {
- timeout = 1;
- break;
- }
- }
-
- if (timeout) {
- if (conf->curr_count > IOT_MIN_THREADS) {
- conf->curr_count--;
- bye = 1;
- gf_log (conf->this->name, GF_LOG_DEBUG,
- "timeout, terminated. conf->curr_count=%d",
- conf->curr_count);
- } else {
- timeout = 0;
- }
- }
-
- stub = __iot_dequeue (conf, &pri, &sleep);
- if (!stub && (sleep.tv_sec || sleep.tv_nsec)) {
- pthread_cond_timedwait(&conf->cond,
- &conf->mutex, &sleep);
- pthread_mutex_unlock(&conf->mutex);
- continue;
- }
+iot_worker(void *data)
+{
+ iot_conf_t *conf = NULL;
+ xlator_t *this = NULL;
+ call_stub_t *stub = NULL;
+ struct timespec sleep_till = {
+ 0,
+ };
+ int ret = 0;
+ int pri = -1;
+ gf_boolean_t bye = _gf_false;
+
+ conf = data;
+ this = conf->this;
+ THIS = this;
+
+ for (;;) {
+ pthread_mutex_lock(&conf->mutex);
+ {
+ if (pri != -1) {
+ conf->ac_iot_count[pri]--;
+ pri = -1;
+ }
+ while (conf->queue_size == 0) {
+ if (conf->down) {
+ bye = _gf_true; /*Avoid sleep*/
+ break;
}
- pthread_mutex_unlock (&conf->mutex);
- if (stub) /* guard against spurious wakeups */
- call_resume (stub);
+ clock_gettime(CLOCK_REALTIME_COARSE, &sleep_till);
+ sleep_till.tv_sec += conf->idle_time;
- if (bye)
- break;
- }
+ conf->sleep_count++;
+ ret = pthread_cond_timedwait(&conf->cond, &conf->mutex,
+ &sleep_till);
+ conf->sleep_count--;
- if (pri != -1) {
- pthread_mutex_lock (&conf->mutex);
- {
- conf->ac_iot_count[pri]--;
+ if (conf->down || ret == ETIMEDOUT) {
+ bye = _gf_true;
+ break;
+ }
+ }
+
+ if (bye) {
+ if (conf->down || conf->curr_count > IOT_MIN_THREADS) {
+ conf->curr_count--;
+ if (conf->curr_count == 0)
+ pthread_cond_broadcast(&conf->cond);
+ gf_msg_debug(conf->this->name, 0,
+ "terminated. "
+ "conf->curr_count=%d",
+ conf->curr_count);
+ } else {
+ bye = _gf_false;
}
- pthread_mutex_unlock (&conf->mutex);
+ }
+
+ if (!bye)
+ stub = __iot_dequeue(conf, &pri);
}
- return NULL;
-}
+ pthread_mutex_unlock(&conf->mutex);
+
+ if (stub) { /* guard against spurious wakeups */
+ if (stub->poison) {
+ gf_log(this->name, GF_LOG_INFO, "Dropping poisoned request %p.",
+ stub);
+ call_stub_destroy(stub);
+ } else {
+ call_resume(stub);
+ }
+ GF_ATOMIC_DEC(conf->stub_cnt);
+ }
+ stub = NULL;
+ if (bye)
+ break;
+ }
+
+ return NULL;
+}
int
-do_iot_schedule (iot_conf_t *conf, call_stub_t *stub, int pri)
+do_iot_schedule(iot_conf_t *conf, call_stub_t *stub, int pri)
{
- int ret = 0;
+ int ret = 0;
- pthread_mutex_lock (&conf->mutex);
- {
- __iot_enqueue (conf, stub, pri);
+ pthread_mutex_lock(&conf->mutex);
+ {
+ __iot_enqueue(conf, stub, pri);
- pthread_cond_signal (&conf->cond);
+ pthread_cond_signal(&conf->cond);
- ret = __iot_workers_scale (conf);
- }
- pthread_mutex_unlock (&conf->mutex);
+ ret = __iot_workers_scale(conf);
+ }
+ pthread_mutex_unlock(&conf->mutex);
- return ret;
+ return ret;
}
-char*
-iot_get_pri_meaning (iot_pri_t pri)
-{
- char *name = NULL;
- switch (pri) {
- case IOT_PRI_HI:
- name = "fast";
- break;
- case IOT_PRI_NORMAL:
- name = "normal";
- break;
- case IOT_PRI_LO:
- name = "slow";
- break;
- case IOT_PRI_LEAST:
- name = "least priority";
- break;
- case IOT_PRI_MAX:
- name = "invalid";
- break;
- }
- return name;
+char *
+iot_get_pri_meaning(gf_fop_pri_t pri)
+{
+ char *name = NULL;
+ switch (pri) {
+ case GF_FOP_PRI_HI:
+ name = "fast";
+ break;
+ case GF_FOP_PRI_NORMAL:
+ name = "normal";
+ break;
+ case GF_FOP_PRI_LO:
+ name = "slow";
+ break;
+ case GF_FOP_PRI_LEAST:
+ name = "least";
+ break;
+ case GF_FOP_PRI_MAX:
+ name = "invalid";
+ break;
+ case GF_FOP_PRI_UNSPEC:
+ name = "unspecified";
+ break;
+ }
+ return name;
}
int
-iot_schedule (call_frame_t *frame, xlator_t *this, call_stub_t *stub)
+iot_schedule(call_frame_t *frame, xlator_t *this, call_stub_t *stub)
{
- int ret = -1;
- iot_pri_t pri = IOT_PRI_MAX - 1;
- iot_conf_t *conf = this->private;
-
- if ((frame->root->pid < GF_CLIENT_PID_MAX) && conf->least_priority) {
- pri = IOT_PRI_LEAST;
- goto out;
- }
-
- switch (stub->fop) {
+ int ret = -1;
+ gf_fop_pri_t pri = GF_FOP_PRI_MAX - 1;
+ iot_conf_t *conf = this->private;
+
+ if ((frame->root->pid < GF_CLIENT_PID_MAX) &&
+ (frame->root->pid != GF_CLIENT_PID_NO_ROOT_SQUASH) &&
+ conf->least_priority) {
+ pri = GF_FOP_PRI_LEAST;
+ goto out;
+ }
+
+ switch (stub->fop) {
case GF_FOP_OPEN:
case GF_FOP_STAT:
case GF_FOP_FSTAT:
@@ -293,8 +312,12 @@ iot_schedule (call_frame_t *frame, xlator_t *this, call_stub_t *stub)
case GF_FOP_STATFS:
case GF_FOP_READDIR:
case GF_FOP_READDIRP:
- pri = IOT_PRI_HI;
- break;
+ case GF_FOP_GETACTIVELK:
+ case GF_FOP_SETACTIVELK:
+ case GF_FOP_ICREATE:
+ case GF_FOP_NAMELINK:
+ pri = GF_FOP_PRI_HI;
+ break;
case GF_FOP_CREATE:
case GF_FOP_FLUSH:
@@ -303,6 +326,7 @@ iot_schedule (call_frame_t *frame, xlator_t *this, call_stub_t *stub)
case GF_FOP_FINODELK:
case GF_FOP_ENTRYLK:
case GF_FOP_FENTRYLK:
+ case GF_FOP_LEASE:
case GF_FOP_UNLINK:
case GF_FOP_SETATTR:
case GF_FOP_FSETATTR:
@@ -318,8 +342,9 @@ iot_schedule (call_frame_t *frame, xlator_t *this, call_stub_t *stub)
case GF_FOP_FSETXATTR:
case GF_FOP_REMOVEXATTR:
case GF_FOP_FREMOVEXATTR:
- pri = IOT_PRI_NORMAL;
- break;
+ case GF_FOP_PUT:
+ pri = GF_FOP_PRI_NORMAL;
+ break;
case GF_FOP_READ:
case GF_FOP_WRITE:
@@ -330,817 +355,1236 @@ iot_schedule (call_frame_t *frame, xlator_t *this, call_stub_t *stub)
case GF_FOP_XATTROP:
case GF_FOP_FXATTROP:
case GF_FOP_RCHECKSUM:
- case GF_FOP_FALLOCATE:
- case GF_FOP_DISCARD:
+ case GF_FOP_FALLOCATE:
+ case GF_FOP_DISCARD:
case GF_FOP_ZEROFILL:
- pri = IOT_PRI_LO;
- break;
+ case GF_FOP_SEEK:
+ pri = GF_FOP_PRI_LO;
+ break;
- case GF_FOP_NULL:
case GF_FOP_FORGET:
case GF_FOP_RELEASE:
case GF_FOP_RELEASEDIR:
case GF_FOP_GETSPEC:
- case GF_FOP_MAXVALUE:
- //fail compilation on missing fop
- //new fop must choose priority.
- break;
- }
+ break;
+ case GF_FOP_IPC:
+ default:
+ return -EINVAL;
+ }
out:
- gf_log (this->name, GF_LOG_DEBUG, "%s scheduled as %s fop",
- gf_fop_list[stub->fop], iot_get_pri_meaning (pri));
- ret = do_iot_schedule (this->private, stub, pri);
- return ret;
+ gf_msg_debug(this->name, 0, "%s scheduled as %s priority fop",
+ gf_fop_list[stub->fop], iot_get_pri_meaning(pri));
+ if (this->private)
+ ret = do_iot_schedule(this->private, stub, pri);
+ return ret;
}
int
-iot_lookup (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
+iot_lookup(call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
{
- IOT_FOP (lookup, frame, this, loc, xdata);
- return 0;
+ IOT_FOP(lookup, frame, this, loc, xdata);
+ return 0;
}
-
int
-iot_setattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
- struct iatt *stbuf, int32_t valid, dict_t *xdata)
+iot_setattr(call_frame_t *frame, xlator_t *this, loc_t *loc, struct iatt *stbuf,
+ int32_t valid, dict_t *xdata)
{
- IOT_FOP (setattr, frame, this, loc, stbuf, valid, xdata);
- return 0;
+ IOT_FOP(setattr, frame, this, loc, stbuf, valid, xdata);
+ return 0;
}
-
int
-iot_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
- struct iatt *stbuf, int32_t valid, dict_t *xdata)
+iot_fsetattr(call_frame_t *frame, xlator_t *this, fd_t *fd, struct iatt *stbuf,
+ int32_t valid, dict_t *xdata)
{
- IOT_FOP (fsetattr, frame, this, fd, stbuf, valid, xdata);
- return 0;
+ IOT_FOP(fsetattr, frame, this, fd, stbuf, valid, xdata);
+ return 0;
}
-
int
-iot_access (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t mask,
- dict_t *xdata)
+iot_access(call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t mask,
+ dict_t *xdata)
{
- IOT_FOP (access, frame, this, loc, mask, xdata);
- return 0;
+ IOT_FOP(access, frame, this, loc, mask, xdata);
+ return 0;
}
-
int
-iot_readlink (call_frame_t *frame, xlator_t *this, loc_t *loc, size_t size, dict_t *xdata)
+iot_readlink(call_frame_t *frame, xlator_t *this, loc_t *loc, size_t size,
+ dict_t *xdata)
{
- IOT_FOP (readlink, frame, this, loc, size, xdata);
- return 0;
+ IOT_FOP(readlink, frame, this, loc, size, xdata);
+ return 0;
}
-
int
-iot_mknod (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
- dev_t rdev, mode_t umask, dict_t *xdata)
+iot_mknod(call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
+ dev_t rdev, mode_t umask, dict_t *xdata)
{
- IOT_FOP (mknod, frame, this, loc, mode, rdev, umask, xdata);
- return 0;
+ IOT_FOP(mknod, frame, this, loc, mode, rdev, umask, xdata);
+ return 0;
}
-
int
-iot_mkdir (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
- mode_t umask, dict_t *xdata)
+iot_mkdir(call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
+ mode_t umask, dict_t *xdata)
{
- IOT_FOP (mkdir, frame, this, loc, mode, umask, xdata);
- return 0;
+ IOT_FOP(mkdir, frame, this, loc, mode, umask, xdata);
+ return 0;
}
-
int
-iot_rmdir (call_frame_t *frame, xlator_t *this, loc_t *loc, int flags, dict_t *xdata)
+iot_rmdir(call_frame_t *frame, xlator_t *this, loc_t *loc, int flags,
+ dict_t *xdata)
{
- IOT_FOP (rmdir, frame, this, loc, flags, xdata);
- return 0;
+ IOT_FOP(rmdir, frame, this, loc, flags, xdata);
+ return 0;
}
-
int
-iot_symlink (call_frame_t *frame, xlator_t *this, const char *linkname,
- loc_t *loc, mode_t umask, dict_t *xdata)
+iot_symlink(call_frame_t *frame, xlator_t *this, const char *linkname,
+ loc_t *loc, mode_t umask, dict_t *xdata)
{
- IOT_FOP (symlink, frame, this, linkname, loc, umask, xdata);
- return 0;
+ IOT_FOP(symlink, frame, this, linkname, loc, umask, xdata);
+ return 0;
}
-
int
-iot_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
- dict_t *xdata)
+iot_rename(call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
+ dict_t *xdata)
{
- IOT_FOP (rename, frame, this, oldloc, newloc, xdata);
- return 0;
+ IOT_FOP(rename, frame, this, oldloc, newloc, xdata);
+ return 0;
}
-
int
-iot_open (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
- fd_t *fd, dict_t *xdata)
+iot_open(call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
+ fd_t *fd, dict_t *xdata)
{
- IOT_FOP (open, frame, this, loc, flags, fd, xdata);
- return 0;
+ IOT_FOP(open, frame, this, loc, flags, fd, xdata);
+ return 0;
}
-
int
-iot_create (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
- mode_t mode, mode_t umask, fd_t *fd, dict_t *xdata)
+iot_create(call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
+ mode_t mode, mode_t umask, fd_t *fd, dict_t *xdata)
{
- IOT_FOP (create, frame, this, loc, flags, mode, umask, fd, xdata);
- return 0;
+ IOT_FOP(create, frame, this, loc, flags, mode, umask, fd, xdata);
+ return 0;
}
-
int
-iot_readv (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
- off_t offset, uint32_t flags, dict_t *xdata)
+iot_put(call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
+ mode_t umask, uint32_t flags, struct iovec *vector, int32_t count,
+ off_t offset, struct iobref *iobref, dict_t *xattr, dict_t *xdata)
{
- IOT_FOP (readv, frame, this, fd, size, offset, flags, xdata);
- return 0;
+ IOT_FOP(put, frame, this, loc, mode, umask, flags, vector, count, offset,
+ iobref, xattr, xdata);
+ return 0;
}
+int
+iot_readv(call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
+ off_t offset, uint32_t flags, dict_t *xdata)
+{
+ IOT_FOP(readv, frame, this, fd, size, offset, flags, xdata);
+ return 0;
+}
int
-iot_flush (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
+iot_flush(call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
{
- IOT_FOP (flush, frame, this, fd, xdata);
- return 0;
+ IOT_FOP(flush, frame, this, fd, xdata);
+ return 0;
}
+int
+iot_fsync(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t datasync,
+ dict_t *xdata)
+{
+ IOT_FOP(fsync, frame, this, fd, datasync, xdata);
+ return 0;
+}
int
-iot_fsync (call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t datasync,
+iot_writev(call_frame_t *frame, xlator_t *this, fd_t *fd, struct iovec *vector,
+ int32_t count, off_t offset, uint32_t flags, struct iobref *iobref,
dict_t *xdata)
{
- IOT_FOP (fsync, frame, this, fd, datasync, xdata);
- return 0;
+ IOT_FOP(writev, frame, this, fd, vector, count, offset, flags, iobref,
+ xdata);
+ return 0;
}
-
int
-iot_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
- struct iovec *vector, int32_t count, off_t offset,
- uint32_t flags, struct iobref *iobref, dict_t *xdata)
+iot_lk(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t cmd,
+ struct gf_flock *flock, dict_t *xdata)
{
- IOT_FOP (writev, frame, this, fd, vector, count, offset,
- flags, iobref, xdata);
- return 0;
+ IOT_FOP(lk, frame, this, fd, cmd, flock, xdata);
+ return 0;
}
-
int
-iot_lk (call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t cmd,
- struct gf_flock *flock, dict_t *xdata)
+iot_stat(call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
{
- IOT_FOP (lk, frame, this, fd, cmd, flock, xdata);
- return 0;
+ IOT_FOP(stat, frame, this, loc, xdata);
+ return 0;
}
-
int
-iot_stat (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
+iot_fstat(call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
{
- IOT_FOP (stat, frame, this, loc, xdata);
- return 0;
+ IOT_FOP(fstat, frame, this, fd, xdata);
+ return 0;
}
-
int
-iot_fstat (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
+iot_truncate(call_frame_t *frame, xlator_t *this, loc_t *loc, off_t offset,
+ dict_t *xdata)
{
- IOT_FOP (fstat, frame, this, fd, xdata);
- return 0;
+ IOT_FOP(truncate, frame, this, loc, offset, xdata);
+ return 0;
}
-
int
-iot_truncate (call_frame_t *frame, xlator_t *this, loc_t *loc, off_t offset,
+iot_ftruncate(call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
dict_t *xdata)
{
- IOT_FOP (truncate, frame, this, loc, offset, xdata);
- return 0;
+ IOT_FOP(ftruncate, frame, this, fd, offset, xdata);
+ return 0;
}
-
int
-iot_ftruncate (call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
- dict_t *xdata)
+iot_unlink(call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t xflag,
+ dict_t *xdata)
{
- IOT_FOP (ftruncate, frame, this, fd, offset, xdata);
- return 0;
+ IOT_FOP(unlink, frame, this, loc, xflag, xdata);
+ return 0;
}
-
-
int
-iot_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t xflag,
- dict_t *xdata)
+iot_link(call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
+ dict_t *xdata)
{
- IOT_FOP (unlink, frame, this, loc, xflag, xdata);
- return 0;
+ IOT_FOP(link, frame, this, oldloc, newloc, xdata);
+ return 0;
}
-
int
-iot_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
- dict_t *xdata)
+iot_opendir(call_frame_t *frame, xlator_t *this, loc_t *loc, fd_t *fd,
+ dict_t *xdata)
{
- IOT_FOP (link, frame, this, oldloc, newloc, xdata);
- return 0;
+ IOT_FOP(opendir, frame, this, loc, fd, xdata);
+ return 0;
}
-
int
-iot_opendir (call_frame_t *frame, xlator_t *this, loc_t *loc, fd_t *fd,
+iot_fsyncdir(call_frame_t *frame, xlator_t *this, fd_t *fd, int datasync,
dict_t *xdata)
{
- IOT_FOP (opendir, frame, this, loc, fd, xdata);
- return 0;
+ IOT_FOP(fsyncdir, frame, this, fd, datasync, xdata);
+ return 0;
}
-
int
-iot_fsyncdir (call_frame_t *frame, xlator_t *this, fd_t *fd, int datasync,
- dict_t *xdata)
+iot_statfs(call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
{
- IOT_FOP (fsyncdir, frame, this, fd, datasync, xdata);
- return 0;
+ IOT_FOP(statfs, frame, this, loc, xdata);
+ return 0;
}
-
int
-iot_statfs (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
+iot_setxattr(call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *dict,
+ int32_t flags, dict_t *xdata)
{
- IOT_FOP (statfs, frame, this, loc, xdata);
- return 0;
+ IOT_FOP(setxattr, frame, this, loc, dict, flags, xdata);
+ return 0;
}
-
int
-iot_setxattr (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *dict,
- int32_t flags, dict_t *xdata)
+iot_getxattr(call_frame_t *frame, xlator_t *this, loc_t *loc, const char *name,
+ dict_t *xdata)
{
- IOT_FOP (setxattr, frame, this, loc, dict, flags, xdata);
+ iot_conf_t *conf = NULL;
+ dict_t *depths = NULL;
+ int i = 0;
+ int32_t op_ret = 0;
+ int32_t op_errno = 0;
+
+ conf = this->private;
+
+ if (name && strcmp(name, IO_THREADS_QUEUE_SIZE_KEY) == 0) {
+ /*
+ * We explicitly do not want a reference count
+ * for this dict in this translator
+ */
+ depths = dict_new();
+ if (!depths) {
+ op_ret = -1;
+ op_errno = ENOMEM;
+ goto unwind_special_getxattr;
+ }
+
+ for (i = 0; i < GF_FOP_PRI_MAX; i++) {
+ if (dict_set_int32(depths, (char *)fop_pri_to_string(i),
+ conf->queue_sizes[i]) != 0) {
+ dict_unref(depths);
+ depths = NULL;
+ goto unwind_special_getxattr;
+ }
+ }
+
+ unwind_special_getxattr:
+ STACK_UNWIND_STRICT(getxattr, frame, op_ret, op_errno, depths, xdata);
+ if (depths)
+ dict_unref(depths);
return 0;
-}
+ }
+ IOT_FOP(getxattr, frame, this, loc, name, xdata);
+ return 0;
+}
int
-iot_getxattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
- const char *name, dict_t *xdata)
+iot_fgetxattr(call_frame_t *frame, xlator_t *this, fd_t *fd, const char *name,
+ dict_t *xdata)
{
- IOT_FOP (getxattr, frame, this, loc, name, xdata);
- return 0;
+ IOT_FOP(fgetxattr, frame, this, fd, name, xdata);
+ return 0;
}
-
int
-iot_fgetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
- const char *name, dict_t *xdata)
+iot_fsetxattr(call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *dict,
+ int32_t flags, dict_t *xdata)
{
- IOT_FOP (fgetxattr, frame, this, fd, name, xdata);
- return 0;
+ IOT_FOP(fsetxattr, frame, this, fd, dict, flags, xdata);
+ return 0;
}
-
int
-iot_fsetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *dict,
- int32_t flags, dict_t *xdata)
+iot_removexattr(call_frame_t *frame, xlator_t *this, loc_t *loc,
+ const char *name, dict_t *xdata)
{
- IOT_FOP (fsetxattr, frame, this, fd, dict, flags, xdata);
- return 0;
+ IOT_FOP(removexattr, frame, this, loc, name, xdata);
+ return 0;
}
-
int
-iot_removexattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
+iot_fremovexattr(call_frame_t *frame, xlator_t *this, fd_t *fd,
const char *name, dict_t *xdata)
{
- IOT_FOP (removexattr, frame, this, loc, name, xdata);
- return 0;
+ IOT_FOP(fremovexattr, frame, this, fd, name, xdata);
+ return 0;
}
int
-iot_fremovexattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
- const char *name, dict_t *xdata)
+iot_readdirp(call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
+ off_t offset, dict_t *xdata)
{
- IOT_FOP (fremovexattr, frame, this, fd, name, xdata);
- return 0;
+ IOT_FOP(readdirp, frame, this, fd, size, offset, xdata);
+ return 0;
}
-
int
-iot_readdirp (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
- off_t offset, dict_t *xdata)
+iot_readdir(call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
+ off_t offset, dict_t *xdata)
{
- IOT_FOP (readdirp, frame, this, fd, size, offset, xdata);
- return 0;
+ IOT_FOP(readdir, frame, this, fd, size, offset, xdata);
+ return 0;
}
-
int
-iot_readdir (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
- off_t offset, dict_t *xdata)
+iot_inodelk(call_frame_t *frame, xlator_t *this, const char *volume, loc_t *loc,
+ int32_t cmd, struct gf_flock *lock, dict_t *xdata)
{
- IOT_FOP (readdir, frame, this, fd, size, offset, xdata);
- return 0;
+ IOT_FOP(inodelk, frame, this, volume, loc, cmd, lock, xdata);
+ return 0;
}
int
-iot_inodelk (call_frame_t *frame, xlator_t *this,
- const char *volume, loc_t *loc, int32_t cmd, struct gf_flock *lock,
- dict_t *xdata)
+iot_finodelk(call_frame_t *frame, xlator_t *this, const char *volume, fd_t *fd,
+ int32_t cmd, struct gf_flock *lock, dict_t *xdata)
{
- IOT_FOP (inodelk, frame, this, volume, loc, cmd, lock, xdata);
- return 0;
+ IOT_FOP(finodelk, frame, this, volume, fd, cmd, lock, xdata);
+ return 0;
}
int
-iot_finodelk (call_frame_t *frame, xlator_t *this,
- const char *volume, fd_t *fd, int32_t cmd, struct gf_flock *lock,
- dict_t *xdata)
+iot_entrylk(call_frame_t *frame, xlator_t *this, const char *volume, loc_t *loc,
+ const char *basename, entrylk_cmd cmd, entrylk_type type,
+ dict_t *xdata)
{
- IOT_FOP (finodelk, frame, this, volume, fd, cmd, lock, xdata);
- return 0;
+ IOT_FOP(entrylk, frame, this, volume, loc, basename, cmd, type, xdata);
+ return 0;
}
int
-iot_entrylk (call_frame_t *frame, xlator_t *this,
- const char *volume, loc_t *loc, const char *basename,
- entrylk_cmd cmd, entrylk_type type, dict_t *xdata)
+iot_fentrylk(call_frame_t *frame, xlator_t *this, const char *volume, fd_t *fd,
+ const char *basename, entrylk_cmd cmd, entrylk_type type,
+ dict_t *xdata)
{
- IOT_FOP (entrylk, frame, this, volume, loc, basename, cmd, type, xdata);
- return 0;
+ IOT_FOP(fentrylk, frame, this, volume, fd, basename, cmd, type, xdata);
+ return 0;
}
int
-iot_fentrylk (call_frame_t *frame, xlator_t *this,
- const char *volume, fd_t *fd, const char *basename,
- entrylk_cmd cmd, entrylk_type type, dict_t *xdata)
+iot_xattrop(call_frame_t *frame, xlator_t *this, loc_t *loc,
+ gf_xattrop_flags_t optype, dict_t *xattr, dict_t *xdata)
{
- IOT_FOP (fentrylk, frame, this, volume, fd, basename, cmd, type, xdata);
- return 0;
+ IOT_FOP(xattrop, frame, this, loc, optype, xattr, xdata);
+ return 0;
}
-
int
-iot_xattrop (call_frame_t *frame, xlator_t *this, loc_t *loc,
+iot_fxattrop(call_frame_t *frame, xlator_t *this, fd_t *fd,
gf_xattrop_flags_t optype, dict_t *xattr, dict_t *xdata)
{
- IOT_FOP (xattrop, frame, this, loc, optype, xattr, xdata);
- return 0;
+ IOT_FOP(fxattrop, frame, this, fd, optype, xattr, xdata);
+ return 0;
}
+int32_t
+iot_rchecksum(call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
+ int32_t len, dict_t *xdata)
+{
+ IOT_FOP(rchecksum, frame, this, fd, offset, len, xdata);
+ return 0;
+}
int
-iot_fxattrop (call_frame_t *frame, xlator_t *this, fd_t *fd,
- gf_xattrop_flags_t optype, dict_t *xattr, dict_t *xdata)
+iot_fallocate(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t mode,
+ off_t offset, size_t len, dict_t *xdata)
{
- IOT_FOP (fxattrop, frame, this, fd, optype, xattr, xdata);
- return 0;
+ IOT_FOP(fallocate, frame, this, fd, mode, offset, len, xdata);
+ return 0;
}
+int
+iot_discard(call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
+ size_t len, dict_t *xdata)
+{
+ IOT_FOP(discard, frame, this, fd, offset, len, xdata);
+ return 0;
+}
-int32_t
-iot_rchecksum (call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
- int32_t len, dict_t *xdata)
+int
+iot_zerofill(call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
+ off_t len, dict_t *xdata)
{
- IOT_FOP (rchecksum, frame, this, fd, offset, len, xdata);
- return 0;
+ IOT_FOP(zerofill, frame, this, fd, offset, len, xdata);
+ return 0;
}
int
-iot_fallocate(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t mode,
- off_t offset, size_t len, dict_t *xdata)
+iot_seek(call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
+ gf_seek_what_t what, dict_t *xdata)
{
- IOT_FOP (fallocate, frame, this, fd, mode, offset, len, xdata);
- return 0;
+ IOT_FOP(seek, frame, this, fd, offset, what, xdata);
+ return 0;
}
int
-iot_discard(call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
- size_t len, dict_t *xdata)
+iot_lease(call_frame_t *frame, xlator_t *this, loc_t *loc,
+ struct gf_lease *lease, dict_t *xdata)
{
- IOT_FOP (discard, frame, this, fd, offset, len, xdata);
- return 0;
+ IOT_FOP(lease, frame, this, loc, lease, xdata);
+ return 0;
}
int
-iot_zerofill(call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
- off_t len, dict_t *xdata)
+iot_getactivelk(call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
{
- IOT_FOP (zerofill, frame, this, fd, offset, len, xdata);
- return 0;
+ IOT_FOP(getactivelk, frame, this, loc, xdata);
+ return 0;
}
+int
+iot_setactivelk(call_frame_t *frame, xlator_t *this, loc_t *loc,
+ lock_migration_info_t *locklist, dict_t *xdata)
+{
+ IOT_FOP(setactivelk, frame, this, loc, locklist, xdata);
+ return 0;
+}
int
-__iot_workers_scale (iot_conf_t *conf)
+__iot_workers_scale(iot_conf_t *conf)
{
- int scale = 0;
- int diff = 0;
- pthread_t thread;
- int ret = 0;
- int i = 0;
+ int scale = 0;
+ int diff = 0;
+ pthread_t thread;
+ int ret = 0;
+ int i = 0;
+
+ for (i = 0; i < GF_FOP_PRI_MAX; i++)
+ scale += min(conf->queue_sizes[i], conf->ac_iot_limit[i]);
+
+ if (scale < IOT_MIN_THREADS)
+ scale = IOT_MIN_THREADS;
+
+ if (scale > conf->max_count)
+ scale = conf->max_count;
+
+ if (conf->curr_count < scale) {
+ diff = scale - conf->curr_count;
+ }
+
+ while (diff) {
+ diff--;
+
+ ret = gf_thread_create(&thread, &conf->w_attr, iot_worker, conf,
+ "iotwr%03hx", conf->curr_count & 0x3ff);
+ if (ret == 0) {
+ pthread_detach(thread);
+ conf->curr_count++;
+ gf_msg_debug(conf->this->name, 0,
+ "scaled threads to %d (queue_size=%d/%d)",
+ conf->curr_count, conf->queue_size, scale);
+ } else {
+ break;
+ }
+ }
- for (i = 0; i < IOT_PRI_MAX; i++)
- scale += min (conf->queue_sizes[i], conf->ac_iot_limit[i]);
+ return diff;
+}
- if (scale < IOT_MIN_THREADS)
- scale = IOT_MIN_THREADS;
+int
+iot_workers_scale(iot_conf_t *conf)
+{
+ int ret = -1;
- if (scale > conf->max_count)
- scale = conf->max_count;
+ if (conf == NULL) {
+ ret = -EINVAL;
+ goto out;
+ }
- if (conf->curr_count < scale) {
- diff = scale - conf->curr_count;
- }
+ pthread_mutex_lock(&conf->mutex);
+ {
+ ret = __iot_workers_scale(conf);
+ }
+ pthread_mutex_unlock(&conf->mutex);
- while (diff) {
- diff --;
+out:
+ return ret;
+}
- ret = gf_thread_create (&thread, &conf->w_attr, iot_worker, conf);
- if (ret == 0) {
- conf->curr_count++;
- gf_log (conf->this->name, GF_LOG_DEBUG,
- "scaled threads to %d (queue_size=%d/%d)",
- conf->curr_count, conf->queue_size, scale);
- } else {
- break;
- }
+int
+set_stack_size(iot_conf_t *conf)
+{
+ int err = 0;
+ size_t stacksize = IOT_THREAD_STACK_SIZE;
+ xlator_t *this = NULL;
+
+ this = THIS;
+
+ err = pthread_attr_init(&conf->w_attr);
+ if (err != 0) {
+ gf_smsg(this->name, GF_LOG_ERROR, err, IO_THREADS_MSG_INIT_FAILED,
+ NULL);
+ return err;
+ }
+
+ err = pthread_attr_setstacksize(&conf->w_attr, stacksize);
+ if (err == EINVAL) {
+ err = pthread_attr_getstacksize(&conf->w_attr, &stacksize);
+ if (!err) {
+ gf_smsg(this->name, GF_LOG_WARNING, 0, IO_THREADS_MSG_SIZE_NOT_SET,
+ "size=%zd", stacksize, NULL);
+ } else {
+ gf_smsg(this->name, GF_LOG_WARNING, 0, IO_THREADS_MSG_SIZE_NOT_SET,
+ NULL);
+ err = 0;
}
+ }
- return diff;
+ conf->stack_size = stacksize;
+ return err;
}
+int32_t
+mem_acct_init(xlator_t *this)
+{
+ int ret = -1;
+
+ if (!this)
+ return ret;
+
+ ret = xlator_mem_acct_init(this, gf_iot_mt_end + 1);
+
+ if (ret != 0) {
+ gf_smsg(this->name, GF_LOG_ERROR, ENOMEM, IO_THREADS_MSG_NO_MEMORY,
+ NULL);
+ return ret;
+ }
+
+ return ret;
+}
int
-iot_workers_scale (iot_conf_t *conf)
+iot_priv_dump(xlator_t *this)
{
- int ret = -1;
+ iot_conf_t *conf = NULL;
+ char key_prefix[GF_DUMP_MAX_BUF_LEN];
+ char key[GF_DUMP_MAX_BUF_LEN];
+ int i = 0;
- if (conf == NULL) {
- ret = -EINVAL;
- goto out;
- }
+ if (!this)
+ return 0;
- pthread_mutex_lock (&conf->mutex);
- {
- ret = __iot_workers_scale (conf);
- }
- pthread_mutex_unlock (&conf->mutex);
+ conf = this->private;
+ if (!conf)
+ return 0;
-out:
- return ret;
+ snprintf(key_prefix, GF_DUMP_MAX_BUF_LEN, "%s.%s", this->type, this->name);
+
+ gf_proc_dump_add_section("%s", key_prefix);
+
+ gf_proc_dump_write("maximum_threads_count", "%d", conf->max_count);
+ gf_proc_dump_write("current_threads_count", "%d", conf->curr_count);
+ gf_proc_dump_write("sleep_count", "%d", conf->sleep_count);
+ gf_proc_dump_write("idle_time", "%d", conf->idle_time);
+ gf_proc_dump_write("stack_size", "%zd", conf->stack_size);
+ gf_proc_dump_write("max_high_priority_threads", "%d",
+ conf->ac_iot_limit[GF_FOP_PRI_HI]);
+ gf_proc_dump_write("max_normal_priority_threads", "%d",
+ conf->ac_iot_limit[GF_FOP_PRI_NORMAL]);
+ gf_proc_dump_write("max_low_priority_threads", "%d",
+ conf->ac_iot_limit[GF_FOP_PRI_LO]);
+ gf_proc_dump_write("max_least_priority_threads", "%d",
+ conf->ac_iot_limit[GF_FOP_PRI_LEAST]);
+ gf_proc_dump_write("current_high_priority_threads", "%d",
+ conf->ac_iot_count[GF_FOP_PRI_HI]);
+ gf_proc_dump_write("current_normal_priority_threads", "%d",
+ conf->ac_iot_count[GF_FOP_PRI_NORMAL]);
+ gf_proc_dump_write("current_low_priority_threads", "%d",
+ conf->ac_iot_count[GF_FOP_PRI_LO]);
+ gf_proc_dump_write("current_least_priority_threads", "%d",
+ conf->ac_iot_count[GF_FOP_PRI_LEAST]);
+ for (i = 0; i < GF_FOP_PRI_MAX; i++) {
+ if (!conf->queue_sizes[i])
+ continue;
+ snprintf(key, sizeof(key), "%s_priority_queue_length",
+ iot_get_pri_meaning(i));
+ gf_proc_dump_write(key, "%d", conf->queue_sizes[i]);
+ }
+
+ return 0;
}
+/*
+ * We use a decay model to keep track and make sure we're not spawning new
+ * threads too often. Each increment adds a large value to a counter, and that
+ * counter keeps ticking back down to zero over a fairly long period. For
+ * example, let's use ONE_WEEK=604800 seconds, and we want to detect when we
+ * have N=3 increments during that time. Thus, our threshold is
+ * (N-1)*ONE_WEEK. To see how it works, look at three examples.
+ *
+ * (a) Two events close together, then one more almost a week later. The
+ * first two events push our counter to 2*ONE_WEEK plus a bit. At the third
+ * event, we decay down to ONE_WEEK plus a bit and then add ONE_WEEK for the
+ * new event, exceeding our threshold.
+ *
+ * (b) One event, then two more almost a week later. At the time of the
+ * second and third events, the counter is already non-zero, so when we add
+ * 2*ONE_WEEK we exceed again.
+ *
+ * (c) Three events, spaced three days apart. At the time of the second
+ * event, we decay down to approxitely ONE_WEEK*4/7 and then add another
+ * ONE_WEEK. At the third event, we decay again down to ONE_WEEK*8/7 and add
+ * another ONE_WEEK, so boom.
+ *
+ * Note that in all three cases if that last event came a day later our counter
+ * would have decayed a bit more and we would *not* exceed our threshold. It's
+ * not exactly the same as a precise "three in one week" limit, but it's very
+ * close and it allows the same kind of tweaking while requiring only constant
+ * space - no arrays of variable length N to allocate or maintain. All we need
+ * (for each queue) is the value plus the time of the last update.
+ */
+
+typedef struct {
+ time_t update_time;
+ uint32_t value;
+} threshold_t;
+/*
+ * Variables so that I can hack these for testing.
+ * TBD: make these tunable?
+ */
+static uint32_t THRESH_SECONDS = 604800;
+static uint32_t THRESH_EVENTS = 3;
+static uint32_t THRESH_LIMIT = 1209600; /* SECONDS * (EVENTS-1) */
+
+static void
+iot_apply_event(xlator_t *this, threshold_t *thresh)
+{
+ time_t delta, now = gf_time();
+
+ /* Refresh for manual testing/debugging. It's cheap. */
+ THRESH_LIMIT = THRESH_SECONDS * (THRESH_EVENTS - 1);
+
+ if (thresh->value && thresh->update_time) {
+ delta = now - thresh->update_time;
+ /* Be careful about underflow. */
+ if (thresh->value <= delta) {
+ thresh->value = 0;
+ } else {
+ thresh->value -= delta;
+ }
+ }
+
+ thresh->value += THRESH_SECONDS;
+ if (thresh->value >= THRESH_LIMIT) {
+ gf_log(this->name, GF_LOG_EMERG, "watchdog firing too often");
+ /*
+ * The default action for SIGTRAP is to dump core, but the fact
+ * that it's distinct from other signals we use means that
+ * there are other possibilities as well (e.g. drop into gdb or
+ * invoke a special handler).
+ */
+ kill(getpid(), SIGTRAP);
+ }
+
+ thresh->update_time = now;
+}
-void
-set_stack_size (iot_conf_t *conf)
-{
- int err = 0;
- size_t stacksize = IOT_THREAD_STACK_SIZE;
- xlator_t *this = NULL;
-
- this = THIS;
-
- pthread_attr_init (&conf->w_attr);
- err = pthread_attr_setstacksize (&conf->w_attr, stacksize);
- if (err == EINVAL) {
- err = pthread_attr_getstacksize (&conf->w_attr, &stacksize);
- if (!err)
- gf_log (this->name, GF_LOG_WARNING,
- "Using default thread stack size %zd",
- stacksize);
- else
- gf_log (this->name, GF_LOG_WARNING,
- "Using default thread stack size");
+static void *
+iot_watchdog(void *arg)
+{
+ xlator_t *this = arg;
+ iot_conf_t *priv = this->private;
+ int i;
+ int bad_times[GF_FOP_PRI_MAX] = {
+ 0,
+ };
+ threshold_t thresholds[GF_FOP_PRI_MAX] = {{
+ 0,
+ }};
+
+ for (;;) {
+ sleep(max(priv->watchdog_secs / 5, 1));
+ pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
+ pthread_mutex_lock(&priv->mutex);
+ for (i = 0; i < GF_FOP_PRI_MAX; ++i) {
+ if (priv->queue_marked[i]) {
+ if (++bad_times[i] >= 5) {
+ gf_log(this->name, GF_LOG_WARNING, "queue %d stalled", i);
+ iot_apply_event(this, &thresholds[i]);
+ /*
+ * We might not get here if the event
+ * put us over our threshold.
+ */
+ ++(priv->ac_iot_limit[i]);
+ bad_times[i] = 0;
+ }
+ } else {
+ bad_times[i] = 0;
+ }
+ priv->queue_marked[i] = (priv->queue_sizes[i] > 0);
}
+ pthread_mutex_unlock(&priv->mutex);
+ pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
+ }
- conf->stack_size = stacksize;
+ /* NOTREACHED */
+ return NULL;
}
+static void
+start_iot_watchdog(xlator_t *this)
+{
+ iot_conf_t *priv = this->private;
+ int ret;
-int32_t
-mem_acct_init (xlator_t *this)
+ if (priv->watchdog_running) {
+ return;
+ }
+
+ ret = pthread_create(&priv->watchdog_thread, NULL, iot_watchdog, this);
+ if (ret == 0) {
+ priv->watchdog_running = _gf_true;
+ } else {
+ gf_log(this->name, GF_LOG_WARNING,
+ "pthread_create(iot_watchdog) failed");
+ }
+}
+
+static void
+stop_iot_watchdog(xlator_t *this)
{
- int ret = -1;
+ iot_conf_t *priv = this->private;
- if (!this)
- return ret;
+ if (!priv->watchdog_running) {
+ return;
+ }
- ret = xlator_mem_acct_init (this, gf_iot_mt_end + 1);
+ if (pthread_cancel(priv->watchdog_thread) != 0) {
+ gf_log(this->name, GF_LOG_WARNING,
+ "pthread_cancel(iot_watchdog) failed");
+ }
- if (ret != 0) {
- gf_log (this->name, GF_LOG_ERROR, "Memory accounting init"
- "failed");
- return ret;
- }
+ if (pthread_join(priv->watchdog_thread, NULL) != 0) {
+ gf_log(this->name, GF_LOG_WARNING, "pthread_join(iot_watchdog) failed");
+ }
- return ret;
+ /* Failure probably means it's already dead. */
+ priv->watchdog_running = _gf_false;
}
int
-iot_priv_dump (xlator_t *this)
+reconfigure(xlator_t *this, dict_t *options)
{
- iot_conf_t *conf = NULL;
- char key_prefix[GF_DUMP_MAX_BUF_LEN];
+ iot_conf_t *conf = NULL;
+ int ret = -1;
- if (!this)
- return 0;
+ conf = this->private;
+ if (!conf)
+ goto out;
- conf = this->private;
- if (!conf)
- return 0;
+ GF_OPTION_RECONF("thread-count", conf->max_count, options, int32, out);
- snprintf (key_prefix, GF_DUMP_MAX_BUF_LEN, "%s.%s", this->type,
- this->name);
+ GF_OPTION_RECONF("high-prio-threads", conf->ac_iot_limit[GF_FOP_PRI_HI],
+ options, int32, out);
- gf_proc_dump_add_section(key_prefix);
+ GF_OPTION_RECONF("normal-prio-threads",
+ conf->ac_iot_limit[GF_FOP_PRI_NORMAL], options, int32,
+ out);
- gf_proc_dump_write("maximum_threads_count", "%d", conf->max_count);
- gf_proc_dump_write("current_threads_count", "%d", conf->curr_count);
- gf_proc_dump_write("sleep_count", "%d", conf->sleep_count);
- gf_proc_dump_write("idle_time", "%d", conf->idle_time);
- gf_proc_dump_write("stack_size", "%zd", conf->stack_size);
- gf_proc_dump_write("high_priority_threads", "%d",
- conf->ac_iot_limit[IOT_PRI_HI]);
- gf_proc_dump_write("normal_priority_threads", "%d",
- conf->ac_iot_limit[IOT_PRI_NORMAL]);
- gf_proc_dump_write("low_priority_threads", "%d",
- conf->ac_iot_limit[IOT_PRI_LO]);
- gf_proc_dump_write("least_priority_threads", "%d",
- conf->ac_iot_limit[IOT_PRI_LEAST]);
+ GF_OPTION_RECONF("low-prio-threads", conf->ac_iot_limit[GF_FOP_PRI_LO],
+ options, int32, out);
- gf_proc_dump_write("cached least rate", "%u",
- conf->throttle.cached_rate);
- gf_proc_dump_write("least rate limit", "%u", conf->throttle.rate_limit);
+ GF_OPTION_RECONF("least-prio-threads", conf->ac_iot_limit[GF_FOP_PRI_LEAST],
+ options, int32, out);
- return 0;
+ GF_OPTION_RECONF("enable-least-priority", conf->least_priority, options,
+ bool, out);
+
+ GF_OPTION_RECONF("cleanup-disconnected-reqs",
+ conf->cleanup_disconnected_reqs, options, bool, out);
+
+ GF_OPTION_RECONF("watchdog-secs", conf->watchdog_secs, options, int32, out);
+
+ GF_OPTION_RECONF("pass-through", this->pass_through, options, bool, out);
+
+ if (conf->watchdog_secs > 0) {
+ start_iot_watchdog(this);
+ } else {
+ stop_iot_watchdog(this);
+ }
+
+ ret = 0;
+out:
+ return ret;
}
int
-reconfigure (xlator_t *this, dict_t *options)
+init(xlator_t *this)
{
- iot_conf_t *conf = NULL;
- int ret = -1;
+ iot_conf_t *conf = NULL;
+ int ret = -1;
+ int i = 0;
- conf = this->private;
- if (!conf)
- goto out;
+ if (!this->children || this->children->next) {
+ gf_smsg("io-threads", GF_LOG_ERROR, 0,
+ IO_THREADS_MSG_XLATOR_CHILD_MISCONFIGURED, NULL);
+ goto out;
+ }
- GF_OPTION_RECONF ("thread-count", conf->max_count, options, int32, out);
+ if (!this->parents) {
+ gf_smsg(this->name, GF_LOG_WARNING, 0, IO_THREADS_MSG_VOL_MISCONFIGURED,
+ NULL);
+ }
- GF_OPTION_RECONF ("high-prio-threads",
- conf->ac_iot_limit[IOT_PRI_HI], options, int32, out);
+ conf = (void *)GF_CALLOC(1, sizeof(*conf), gf_iot_mt_iot_conf_t);
+ if (conf == NULL) {
+ gf_smsg(this->name, GF_LOG_ERROR, ENOMEM, IO_THREADS_MSG_OUT_OF_MEMORY,
+ NULL);
+ goto out;
+ }
- GF_OPTION_RECONF ("normal-prio-threads",
- conf->ac_iot_limit[IOT_PRI_NORMAL], options, int32,
- out);
+ if ((ret = pthread_cond_init(&conf->cond, NULL)) != 0) {
+ gf_smsg(this->name, GF_LOG_ERROR, 0, IO_THREADS_MSG_PTHREAD_INIT_FAILED,
+ "pthread_cond_init ret=%d", ret, NULL);
+ goto out;
+ }
+ conf->cond_inited = _gf_true;
- GF_OPTION_RECONF ("low-prio-threads",
- conf->ac_iot_limit[IOT_PRI_LO], options, int32, out);
+ if ((ret = pthread_mutex_init(&conf->mutex, NULL)) != 0) {
+ gf_smsg(this->name, GF_LOG_ERROR, 0, IO_THREADS_MSG_PTHREAD_INIT_FAILED,
+ "pthread_mutex_init ret=%d", ret, NULL);
+ goto out;
+ }
+ conf->mutex_inited = _gf_true;
- GF_OPTION_RECONF ("least-prio-threads",
- conf->ac_iot_limit[IOT_PRI_LEAST], options, int32,
- out);
- GF_OPTION_RECONF ("enable-least-priority", conf->least_priority,
- options, bool, out);
+ ret = set_stack_size(conf);
- GF_OPTION_RECONF("least-rate-limit", conf->throttle.rate_limit, options,
- int32, out);
+ if (ret != 0)
+ goto out;
- ret = 0;
-out:
- return ret;
-}
+ ret = -1;
+ GF_OPTION_INIT("thread-count", conf->max_count, int32, out);
-int
-init (xlator_t *this)
-{
- iot_conf_t *conf = NULL;
- int ret = -1;
- int i = 0;
-
- if (!this->children || this->children->next) {
- gf_log ("io-threads", GF_LOG_ERROR,
- "FATAL: iot not configured with exactly one child");
- goto out;
- }
-
- if (!this->parents) {
- gf_log (this->name, GF_LOG_WARNING,
- "dangling volume. check volfile ");
- }
-
- conf = (void *) GF_CALLOC (1, sizeof (*conf),
- gf_iot_mt_iot_conf_t);
- if (conf == NULL) {
- gf_log (this->name, GF_LOG_ERROR,
- "out of memory");
- goto out;
- }
+ GF_OPTION_INIT("high-prio-threads", conf->ac_iot_limit[GF_FOP_PRI_HI],
+ int32, out);
- if ((ret = pthread_cond_init(&conf->cond, NULL)) != 0) {
- gf_log (this->name, GF_LOG_ERROR,
- "pthread_cond_init failed (%d)", ret);
- goto out;
- }
+ GF_OPTION_INIT("normal-prio-threads", conf->ac_iot_limit[GF_FOP_PRI_NORMAL],
+ int32, out);
- if ((ret = pthread_mutex_init(&conf->mutex, NULL)) != 0) {
- gf_log (this->name, GF_LOG_ERROR,
- "pthread_mutex_init failed (%d)", ret);
- goto out;
- }
+ GF_OPTION_INIT("low-prio-threads", conf->ac_iot_limit[GF_FOP_PRI_LO], int32,
+ out);
- set_stack_size (conf);
+ GF_OPTION_INIT("least-prio-threads", conf->ac_iot_limit[GF_FOP_PRI_LEAST],
+ int32, out);
- GF_OPTION_INIT ("thread-count", conf->max_count, int32, out);
+ GF_OPTION_INIT("idle-time", conf->idle_time, int32, out);
- GF_OPTION_INIT ("high-prio-threads",
- conf->ac_iot_limit[IOT_PRI_HI], int32, out);
+ GF_OPTION_INIT("enable-least-priority", conf->least_priority, bool, out);
- GF_OPTION_INIT ("normal-prio-threads",
- conf->ac_iot_limit[IOT_PRI_NORMAL], int32, out);
+ GF_OPTION_INIT("cleanup-disconnected-reqs", conf->cleanup_disconnected_reqs,
+ bool, out);
- GF_OPTION_INIT ("low-prio-threads",
- conf->ac_iot_limit[IOT_PRI_LO], int32, out);
+ GF_OPTION_INIT("pass-through", this->pass_through, bool, out);
- GF_OPTION_INIT ("least-prio-threads",
- conf->ac_iot_limit[IOT_PRI_LEAST], int32, out);
+ conf->this = this;
+ GF_ATOMIC_INIT(conf->stub_cnt, 0);
- GF_OPTION_INIT ("idle-time", conf->idle_time, int32, out);
- GF_OPTION_INIT ("enable-least-priority", conf->least_priority,
- bool, out);
+ for (i = 0; i < GF_FOP_PRI_MAX; i++) {
+ INIT_LIST_HEAD(&conf->clients[i]);
+ INIT_LIST_HEAD(&conf->no_client[i].clients);
+ INIT_LIST_HEAD(&conf->no_client[i].reqs);
+ }
- GF_OPTION_INIT("least-rate-limit", conf->throttle.rate_limit, int32,
- out);
- if ((ret = pthread_mutex_init(&conf->throttle.lock, NULL)) != 0) {
- gf_log (this->name, GF_LOG_ERROR,
- "pthread_mutex_init failed (%d)", ret);
- goto out;
- }
-
- conf->this = this;
+ if (!this->pass_through) {
+ ret = iot_workers_scale(conf);
- for (i = 0; i < IOT_PRI_MAX; i++) {
- INIT_LIST_HEAD (&conf->reqs[i]);
+ if (ret == -1) {
+ gf_smsg(this->name, GF_LOG_ERROR, 0,
+ IO_THREADS_MSG_WORKER_THREAD_INIT_FAILED, NULL);
+ goto out;
}
+ }
- ret = iot_workers_scale (conf);
+ this->private = conf;
- if (ret == -1) {
- gf_log (this->name, GF_LOG_ERROR,
- "cannot initialize worker threads, exiting init");
- goto out;
- }
+ conf->watchdog_secs = 0;
+ GF_OPTION_INIT("watchdog-secs", conf->watchdog_secs, int32, out);
+ if (conf->watchdog_secs > 0) {
+ start_iot_watchdog(this);
+ }
- this->private = conf;
- ret = 0;
+ ret = 0;
out:
- if (ret)
- GF_FREE (conf);
+ if (ret)
+ GF_FREE(conf);
- return ret;
+ return ret;
}
+static void
+iot_exit_threads(iot_conf_t *conf)
+{
+ pthread_mutex_lock(&conf->mutex);
+ {
+ conf->down = _gf_true;
+ /*Let all the threads know that xl is going down*/
+ pthread_cond_broadcast(&conf->cond);
+ while (conf->curr_count) /*Wait for threads to exit*/
+ pthread_cond_wait(&conf->cond, &conf->mutex);
+ }
+ pthread_mutex_unlock(&conf->mutex);
+}
+
+int
+notify(xlator_t *this, int32_t event, void *data, ...)
+{
+ iot_conf_t *conf = this->private;
+ xlator_t *victim = data;
+ uint64_t stub_cnt = 0;
+ struct timespec sleep_till = {
+ 0,
+ };
+
+ if (GF_EVENT_PARENT_DOWN == event) {
+ if (victim->cleanup_starting) {
+ /* Wait for draining stub from queue before notify PARENT_DOWN */
+ stub_cnt = GF_ATOMIC_GET(conf->stub_cnt);
+ if (stub_cnt) {
+ timespec_now_realtime(&sleep_till);
+ sleep_till.tv_sec += 1;
+ pthread_mutex_lock(&conf->mutex);
+ {
+ while (stub_cnt) {
+ (void)pthread_cond_timedwait(&conf->cond, &conf->mutex,
+ &sleep_till);
+ stub_cnt = GF_ATOMIC_GET(conf->stub_cnt);
+ }
+ }
+ pthread_mutex_unlock(&conf->mutex);
+ }
+
+ gf_log(this->name, GF_LOG_INFO,
+ "Notify GF_EVENT_PARENT_DOWN for brick %s", victim->name);
+ } else {
+ iot_exit_threads(conf);
+ }
+ }
+
+ if (GF_EVENT_CHILD_DOWN == event) {
+ if (victim->cleanup_starting) {
+ iot_exit_threads(conf);
+ gf_log(this->name, GF_LOG_INFO,
+ "Notify GF_EVENT_CHILD_DOWN for brick %s", victim->name);
+ }
+ }
+
+ default_notify(this, event, data);
+
+ return 0;
+}
void
-fini (xlator_t *this)
+fini(xlator_t *this)
{
- iot_conf_t *conf = this->private;
+ iot_conf_t *conf = this->private;
+
+ if (!conf)
+ return;
+
+ if (conf->mutex_inited && conf->cond_inited)
+ iot_exit_threads(conf);
+
+ if (conf->cond_inited)
+ pthread_cond_destroy(&conf->cond);
+
+ if (conf->mutex_inited)
+ pthread_mutex_destroy(&conf->mutex);
+
+ stop_iot_watchdog(this);
- GF_FREE (conf);
+ GF_FREE(conf);
- this->private = NULL;
- return;
+ this->private = NULL;
+ return;
+}
+
+int
+iot_client_destroy(xlator_t *this, client_t *client)
+{
+ void *tmp = NULL;
+
+ if (client_ctx_del(client, this, &tmp) == 0) {
+ GF_FREE(tmp);
+ }
+
+ return 0;
+}
+
+static int
+iot_disconnect_cbk(xlator_t *this, client_t *client)
+{
+ int i;
+ call_stub_t *curr;
+ call_stub_t *next;
+ iot_conf_t *conf = this->private;
+ iot_client_ctx_t *ctx;
+
+ if (!conf || !conf->cleanup_disconnected_reqs) {
+ goto out;
+ }
+
+ pthread_mutex_lock(&conf->mutex);
+ for (i = 0; i < GF_FOP_PRI_MAX; i++) {
+ ctx = &conf->no_client[i];
+ list_for_each_entry_safe(curr, next, &ctx->reqs, list)
+ {
+ if (curr->frame->root->client != client) {
+ continue;
+ }
+ gf_log(this->name, GF_LOG_INFO,
+ "poisoning %s fop at %p for client %s",
+ gf_fop_list[curr->fop], curr, client->client_uid);
+ curr->poison = _gf_true;
+ }
+ }
+ pthread_mutex_unlock(&conf->mutex);
+
+out:
+ return 0;
}
struct xlator_dumpops dumpops = {
- .priv = iot_priv_dump,
+ .priv = iot_priv_dump,
};
struct xlator_fops fops = {
- .open = iot_open,
- .create = iot_create,
- .readv = iot_readv,
- .writev = iot_writev,
- .flush = iot_flush,
- .fsync = iot_fsync,
- .lk = iot_lk,
- .stat = iot_stat,
- .fstat = iot_fstat,
- .truncate = iot_truncate,
- .ftruncate = iot_ftruncate,
- .unlink = iot_unlink,
- .lookup = iot_lookup,
- .setattr = iot_setattr,
- .fsetattr = iot_fsetattr,
- .access = iot_access,
- .readlink = iot_readlink,
- .mknod = iot_mknod,
- .mkdir = iot_mkdir,
- .rmdir = iot_rmdir,
- .symlink = iot_symlink,
- .rename = iot_rename,
- .link = iot_link,
- .opendir = iot_opendir,
- .fsyncdir = iot_fsyncdir,
- .statfs = iot_statfs,
- .setxattr = iot_setxattr,
- .getxattr = iot_getxattr,
- .fgetxattr = iot_fgetxattr,
- .fsetxattr = iot_fsetxattr,
- .removexattr = iot_removexattr,
- .fremovexattr = iot_fremovexattr,
- .readdir = iot_readdir,
- .readdirp = iot_readdirp,
- .inodelk = iot_inodelk,
- .finodelk = iot_finodelk,
- .entrylk = iot_entrylk,
- .fentrylk = iot_fentrylk,
- .xattrop = iot_xattrop,
- .fxattrop = iot_fxattrop,
- .rchecksum = iot_rchecksum,
- .fallocate = iot_fallocate,
- .discard = iot_discard,
- .zerofill = iot_zerofill,
+ .open = iot_open,
+ .create = iot_create,
+ .readv = iot_readv,
+ .writev = iot_writev,
+ .flush = iot_flush,
+ .fsync = iot_fsync,
+ .lk = iot_lk,
+ .stat = iot_stat,
+ .fstat = iot_fstat,
+ .truncate = iot_truncate,
+ .ftruncate = iot_ftruncate,
+ .unlink = iot_unlink,
+ .lookup = iot_lookup,
+ .setattr = iot_setattr,
+ .fsetattr = iot_fsetattr,
+ .access = iot_access,
+ .readlink = iot_readlink,
+ .mknod = iot_mknod,
+ .mkdir = iot_mkdir,
+ .rmdir = iot_rmdir,
+ .symlink = iot_symlink,
+ .rename = iot_rename,
+ .link = iot_link,
+ .opendir = iot_opendir,
+ .fsyncdir = iot_fsyncdir,
+ .statfs = iot_statfs,
+ .setxattr = iot_setxattr,
+ .getxattr = iot_getxattr,
+ .fgetxattr = iot_fgetxattr,
+ .fsetxattr = iot_fsetxattr,
+ .removexattr = iot_removexattr,
+ .fremovexattr = iot_fremovexattr,
+ .readdir = iot_readdir,
+ .readdirp = iot_readdirp,
+ .inodelk = iot_inodelk,
+ .finodelk = iot_finodelk,
+ .entrylk = iot_entrylk,
+ .fentrylk = iot_fentrylk,
+ .xattrop = iot_xattrop,
+ .fxattrop = iot_fxattrop,
+ .rchecksum = iot_rchecksum,
+ .fallocate = iot_fallocate,
+ .discard = iot_discard,
+ .zerofill = iot_zerofill,
+ .seek = iot_seek,
+ .lease = iot_lease,
+ .getactivelk = iot_getactivelk,
+ .setactivelk = iot_setactivelk,
+ .put = iot_put,
};
-struct xlator_cbks cbks;
+struct xlator_cbks cbks = {
+ .client_destroy = iot_client_destroy,
+ .client_disconnect = iot_disconnect_cbk,
+};
struct volume_options options[] = {
- { .key = {"thread-count"},
- .type = GF_OPTION_TYPE_INT,
- .min = IOT_MIN_THREADS,
- .max = IOT_MAX_THREADS,
- .default_value = "16",
- .description = "Number of threads in IO threads translator which "
- "perform concurrent IO operations"
-
- },
- { .key = {"high-prio-threads"},
- .type = GF_OPTION_TYPE_INT,
- .min = IOT_MIN_THREADS,
- .max = IOT_MAX_THREADS,
- .default_value = "16",
- .description = "Max number of threads in IO threads translator which "
- "perform high priority IO operations at a given time"
-
- },
- { .key = {"normal-prio-threads"},
- .type = GF_OPTION_TYPE_INT,
- .min = IOT_MIN_THREADS,
- .max = IOT_MAX_THREADS,
- .default_value = "16",
- .description = "Max number of threads in IO threads translator which "
- "perform normal priority IO operations at a given time"
-
- },
- { .key = {"low-prio-threads"},
- .type = GF_OPTION_TYPE_INT,
- .min = IOT_MIN_THREADS,
- .max = IOT_MAX_THREADS,
- .default_value = "16",
- .description = "Max number of threads in IO threads translator which "
- "perform low priority IO operations at a given time"
-
- },
- { .key = {"least-prio-threads"},
- .type = GF_OPTION_TYPE_INT,
- .min = IOT_MIN_THREADS,
- .max = IOT_MAX_THREADS,
- .default_value = "1",
- .description = "Max number of threads in IO threads translator which "
- "perform least priority IO operations at a given time"
- },
- { .key = {"enable-least-priority"},
- .type = GF_OPTION_TYPE_BOOL,
- .default_value = "on",
- .description = "Enable/Disable least priority"
- },
- {.key = {"idle-time"},
- .type = GF_OPTION_TYPE_INT,
- .min = 1,
- .max = 0x7fffffff,
- .default_value = "120",
- },
- {.key = {"least-rate-limit"},
- .type = GF_OPTION_TYPE_INT,
- .min = 0,
- .max = INT_MAX,
- .default_value = "0",
- .description = "Max number of least priority operations to handle "
- "per-second"
- },
- { .key = {NULL},
- },
+ {.key = {"thread-count"},
+ .type = GF_OPTION_TYPE_INT,
+ .min = IOT_MIN_THREADS,
+ .max = IOT_MAX_THREADS,
+ .default_value = "16",
+ .op_version = {1},
+ .flags = OPT_FLAG_SETTABLE | OPT_FLAG_DOC,
+ .tags = {"io-threads"},
+ /*.option = "thread-count"*/
+ .description = "Number of threads in IO threads translator which "
+ "perform concurrent IO operations"
+
+ },
+ {.key = {"high-prio-threads"},
+ .type = GF_OPTION_TYPE_INT,
+ .min = IOT_MIN_THREADS,
+ .max = IOT_MAX_THREADS,
+ .default_value = "16",
+ .op_version = {1},
+ .flags = OPT_FLAG_SETTABLE | OPT_FLAG_DOC,
+ .tags = {"io-threads"},
+ .description = "Max number of threads in IO threads translator which "
+ "perform high priority IO operations at a given time"
+
+ },
+ {.key = {"normal-prio-threads"},
+ .type = GF_OPTION_TYPE_INT,
+ .min = IOT_MIN_THREADS,
+ .max = IOT_MAX_THREADS,
+ .default_value = "16",
+ .op_version = {1},
+ .flags = OPT_FLAG_SETTABLE | OPT_FLAG_DOC,
+ .tags = {"io-threads"},
+ .description = "Max number of threads in IO threads translator which "
+ "perform normal priority IO operations at a given time"
+
+ },
+ {.key = {"low-prio-threads"},
+ .type = GF_OPTION_TYPE_INT,
+ .min = IOT_MIN_THREADS,
+ .max = IOT_MAX_THREADS,
+ .default_value = "16",
+ .op_version = {1},
+ .flags = OPT_FLAG_SETTABLE | OPT_FLAG_DOC,
+ .tags = {"io-threads"},
+ .description = "Max number of threads in IO threads translator which "
+ "perform low priority IO operations at a given time"
+
+ },
+ {.key = {"least-prio-threads"},
+ .type = GF_OPTION_TYPE_INT,
+ .min = IOT_MIN_THREADS,
+ .max = IOT_MAX_THREADS,
+ .default_value = "1",
+ .op_version = {1},
+ .flags = OPT_FLAG_SETTABLE | OPT_FLAG_DOC,
+ .tags = {"io-threads"},
+ .description = "Max number of threads in IO threads translator which "
+ "perform least priority IO operations at a given time"},
+ {.key = {"enable-least-priority"},
+ .type = GF_OPTION_TYPE_BOOL,
+ .default_value = SITE_H_ENABLE_LEAST_PRIORITY,
+ .op_version = {1},
+ .flags = OPT_FLAG_SETTABLE | OPT_FLAG_DOC,
+ .tags = {"io-threads"},
+ .description = "Enable/Disable least priority"},
+ {
+ .key = {"idle-time"},
+ .type = GF_OPTION_TYPE_INT,
+ .min = 1,
+ .max = 0x7fffffff,
+ .default_value = "120",
+ },
+ {.key = {"watchdog-secs"},
+ .type = GF_OPTION_TYPE_INT,
+ .min = 0,
+ .default_value = 0,
+ .op_version = {GD_OP_VERSION_4_1_0},
+ .flags = OPT_FLAG_SETTABLE | OPT_FLAG_DOC,
+ .tags = {"io-threads"},
+ .description = "Number of seconds a queue must be stalled before "
+ "starting an 'emergency' thread."},
+ {.key = {"cleanup-disconnected-reqs"},
+ .type = GF_OPTION_TYPE_BOOL,
+ .default_value = "off",
+ .op_version = {GD_OP_VERSION_4_1_0},
+ .flags = OPT_FLAG_SETTABLE | OPT_FLAG_DOC | OPT_FLAG_CLIENT_OPT,
+ .tags = {"io-threads"},
+ .description = "'Poison' queued requests when a client disconnects"},
+ {.key = {"pass-through"},
+ .type = GF_OPTION_TYPE_BOOL,
+ .default_value = "false",
+ .op_version = {GD_OP_VERSION_4_1_0},
+ .flags = OPT_FLAG_SETTABLE | OPT_FLAG_DOC | OPT_FLAG_CLIENT_OPT,
+ .tags = {"io-threads"},
+ .description = "Enable/Disable io threads translator"},
+ {
+ .key = {NULL},
+ },
+};
+
+xlator_api_t xlator_api = {
+ .init = init,
+ .fini = fini,
+ .notify = notify,
+ .reconfigure = reconfigure,
+ .mem_acct_init = mem_acct_init,
+ .op_version = {1}, /* Present from the initial version */
+ .dumpops = &dumpops,
+ .fops = &fops,
+ .cbks = &cbks,
+ .options = options,
+ .identifier = "io-threads",
+ .category = GF_MAINTAINED,
};
generated by cgit (git 2.34.1) at 2025-12-28 15:22:46 +0000