[Pvfs2-cvs] commit by slang in pvfs2/src/common/mgmt: pint-mgmt.c
pint-mgmt.h pint-op.h pint-queue.c pint-worker-queues.c
pint-worker-threaded-queues.c pint-worker.h
CVS commit program
cvs at parl.clemson.edu
Tue Aug 26 10:21:46 EDT 2008
Update of /projects/cvsroot/pvfs2/src/common/mgmt
In directory parlweb1:/tmp/cvs-serv31901/src/common/mgmt
Modified Files:
Tag: directio-branch
pint-mgmt.c pint-mgmt.h pint-op.h pint-queue.c
pint-worker-queues.c pint-worker-threaded-queues.c
pint-worker.h
Log Message:
fixes for cancel.
Index: pint-mgmt.c
===================================================================
RCS file: /projects/cvsroot/pvfs2/src/common/mgmt/Attic/pint-mgmt.c,v
diff -p -u -r1.1.2.1 -r1.1.2.2
--- pint-mgmt.c 29 Jul 2008 22:48:56 -0000 1.1.2.1
+++ pint-mgmt.c 26 Aug 2008 14:21:46 -0000 1.1.2.2
@@ -1071,7 +1071,7 @@ int PINT_manager_test_op(PINT_manager_t
if(worker->impl->do_work)
{
ret = worker->impl->do_work(
- manager, &worker->inst, context, op_id, microsecs);
+ manager, &worker->inst, context, &entry->op, microsecs);
if(ret < 0)
{
return ret;
@@ -1110,6 +1110,55 @@ int PINT_manager_wait(PINT_manager_t man
return PINT_manager_wait_context(manager, manager->context, microsecs);
}
+/* Cancel an individual operation. This function attempts to cancel
+ * a posted operation.
+ *
+ * @param manager the manager the operation was posted to
+ * @param op_id the operation id to test on
+ *
+ * @return 0 on successful cancellation, otherwise, return -PVFS_error on error
+ */
+int PINT_manager_cancel(PINT_manager_t manager,
+ PINT_op_id op_id)
+{
+ int ret;
+ struct PINT_worker_s *worker;
+ struct PINT_op_entry *entry;
+ PINT_context_id context;
+ int (*cancel_impl)(struct PINT_manager_s *,
+ PINT_worker_inst *,
+ PINT_queue_id,
+ PINT_operation_t *);
+
+ entry = id_gen_fast_lookup(op_id);
+ if(!entry)
+ {
+ return -PVFS_EINVAL;
+ }
+ worker = id_gen_fast_lookup(entry->wq_id);
+ context = entry->ctx_id;
+
+ /* don't use the entry for anything else since it might get freed
+ * by the worker calling complete_op
+ */
+ entry = NULL;
+
+ gen_mutex_lock(&manager->mutex);
+ cancel_impl = worker->impl->cancel;
+ gen_mutex_unlock(&manager->mutex);
+
+ if(cancel_impl)
+ {
+ ret = cancel_impl(
+ manager, &worker->inst, context, &entry->op);
+ if(ret < 0)
+ {
+ return ret;
+ }
+ }
+ return 0;
+}
+
int PINT_manager_wait_context(PINT_manager_t manager,
PINT_context_id context,
int microsecs)
@@ -1171,7 +1220,8 @@ int PINT_manager_wait_op(PINT_manager_t
if(worker->impl->do_work)
{
ret = worker->impl->do_work(
- manager, &worker->inst, entry->ctx_id, op_id, msecs_remaining);
+ manager, &worker->inst, entry->ctx_id,
+ &entry->op, msecs_remaining);
if(ret < 0)
{
return ret;
@@ -1309,7 +1359,7 @@ void PINT_manager_event_handler_add(PINT
* @return 0 on successful servicing. Right now this function always succeeds.
*/
int PINT_manager_service_op(PINT_manager_t manager,
- PINT_operation_t *op,
+ PINT_operation_t *op,
int *service_time,
int *error)
{
Index: pint-mgmt.h
===================================================================
RCS file: /projects/cvsroot/pvfs2/src/common/mgmt/Attic/pint-mgmt.h,v
diff -p -u -r1.1.2.1 -r1.1.2.2
--- pint-mgmt.h 29 Jul 2008 22:48:56 -0000 1.1.2.1
+++ pint-mgmt.h 26 Aug 2008 14:21:46 -0000 1.1.2.2
@@ -107,6 +107,9 @@ int PINT_manager_ctx_post(PINT_manager_t
PVFS_hint *hint,
PVFS_id_gen_t queue_worker_id);
+int PINT_manager_cancel(PINT_manager_t manager,
+ PINT_op_id op_id);
+
typedef int (*PINT_worker_mapping_callout) (PINT_manager_t manager,
PINT_service_callout callout,
Index: pint-op.h
===================================================================
RCS file: /projects/cvsroot/pvfs2/src/common/mgmt/Attic/pint-op.h,v
diff -p -u -r1.1.2.1 -r1.1.2.2
--- pint-op.h 29 Jul 2008 22:48:56 -0000 1.1.2.1
+++ pint-op.h 26 Aug 2008 14:21:46 -0000 1.1.2.2
@@ -33,6 +33,7 @@ typedef struct
{
PINT_op_id id;
PINT_service_callout operation;
+ PINT_service_callout cancel;
void *operation_ptr;
PVFS_hint hint;
Index: pint-queue.c
===================================================================
RCS file: /projects/cvsroot/pvfs2/src/common/mgmt/Attic/pint-queue.c,v
diff -p -u -r1.1.2.2 -r1.1.2.3
--- pint-queue.c 14 Aug 2008 22:11:16 -0000 1.1.2.2
+++ pint-queue.c 26 Aug 2008 14:21:46 -0000 1.1.2.3
@@ -340,6 +340,12 @@ int PINT_queue_remove(PINT_queue_id queu
struct PINT_queue_trigger *trigger;
PINT_queue_entry_t *e, *tmp;
struct timeval now;
+ int ret = -PVFS_ENOENT;
+
+ if(entry->link.prev == NULL)
+ {
+ return -PVFS_ENOENT;
+ }
queue = id_gen_fast_lookup(queue_id);
@@ -363,6 +369,7 @@ int PINT_queue_remove(PINT_queue_id queu
PINT_queue_update_stats(
queue,
PINT_util_get_timeval_diff(&e->timestamp, &now));
+ ret = 0;
break;
}
}
@@ -383,7 +390,7 @@ int PINT_queue_remove(PINT_queue_id queu
gen_mutex_unlock(&queue->mutex);
- return 0;
+ return ret;
}
int PINT_queue_search_and_remove(PINT_queue_id queue_id,
Index: pint-worker-queues.c
===================================================================
RCS file: /projects/cvsroot/pvfs2/src/common/mgmt/Attic/pint-worker-queues.c,v
diff -p -u -r1.1.2.1 -r1.1.2.2
--- pint-worker-queues.c 29 Jul 2008 22:48:57 -0000 1.1.2.1
+++ pint-worker-queues.c 26 Aug 2008 14:21:46 -0000 1.1.2.2
@@ -192,7 +192,7 @@ static int queues_post(struct PINT_manag
static int queues_do_work(struct PINT_manager_s *manager,
PINT_worker_inst *inst,
PINT_context_id context_id,
- PINT_op_id op_id,
+ PINT_operation_t *op,
int microsecs)
{
struct PINT_worker_queues_s *w;
@@ -200,7 +200,6 @@ static int queues_do_work(struct PINT_ma
struct PINT_queue_s *queue;
int count;
int i, j, ret;
- PINT_operation_t *op = NULL;
PINT_queue_entry_t *qentry;
int service_time, error;
@@ -216,7 +215,7 @@ static int queues_do_work(struct PINT_ma
return 0;
}
- if(op_id != 0)
+ if(op->id != 0)
{
/* find the op in one of the queues */
qlist_for_each_entry(queue, &w->queues, link)
@@ -224,7 +223,7 @@ static int queues_do_work(struct PINT_ma
ret = PINT_queue_search_and_remove(
queue->id,
PINT_op_queue_find_op_id_callback,
- &op_id,
+ &op->id,
&qentry);
if(ret == -PVFS_ENOENT)
{
@@ -363,7 +362,8 @@ struct PINT_worker_impl PINT_worker_queu
queues_queue_add,
queues_queue_remove,
queues_post,
- queues_do_work
+ queues_do_work,
+ NULL
};
/*
Index: pint-worker-threaded-queues.c
===================================================================
RCS file: /projects/cvsroot/pvfs2/src/common/mgmt/Attic/pint-worker-threaded-queues.c,v
diff -p -u -r1.1.2.2 -r1.1.2.3
--- pint-worker-threaded-queues.c 14 Aug 2008 17:51:09 -0000 1.1.2.2
+++ pint-worker-threaded-queues.c 26 Aug 2008 14:21:46 -0000 1.1.2.3
@@ -236,6 +236,53 @@ static int threaded_queues_post(struct P
return PINT_MGMT_OP_POSTED;
}
+static int threaded_queues_cancel(struct PINT_manager_s *manager,
+ PINT_worker_inst *inst,
+ PINT_queue_id queue_id,
+ PINT_operation_t *op)
+{
+ struct PINT_worker_threaded_queues_s *w;
+ struct PINT_queue_s *queue;
+
+ w = &inst->threaded;
+
+ gen_mutex_lock(&w->mutex);
+
+ /* if the queue_id matches the worker_id, then assume that there's
+ * only one queue maintained by this worker and use that
+ */
+ if(queue_id == 0)
+ {
+ /* a dirty hack to check that the list of queues only has one element */
+ if(w->queues.next->next != NULL)
+ {
+ gossip_err("%s: no queue id was specified and there's more than "
+ "one queue being managed by this worker\n", __func__);
+ gen_mutex_unlock(&w->mutex);
+ return -PVFS_EINVAL;
+ }
+
+ /* there must be only one queue, so just use that */
+ queue = qlist_entry(&w->queues.next, struct PINT_queue_s, link);
+ queue_id = queue->id;
+ }
+ else
+ {
+ /* its not the worker id, so it must be an id for one of our queues */
+ queue = id_gen_fast_lookup(queue_id);
+
+ /* verify that this is a queue id we know about */
+ if(!qlist_exists(&w->queues, &queue->link) &&
+ !qlist_exists(&w->inuse_queues, &queue->link))
+ {
+ gen_mutex_unlock(&w->mutex);
+ return -PVFS_EINVAL;
+ }
+ }
+
+ return PINT_queue_remove(queue_id, &op->qentry);
+}
+
struct PINT_worker_impl PINT_worker_threaded_queues_impl =
{
"THREADED",
@@ -248,7 +295,9 @@ struct PINT_worker_impl PINT_worker_thre
/* the threaded queues impl doesn't implement the do_work callback
* because work is done in the threads
*/
- NULL
+ NULL,
+
+ threaded_queues_cancel
};
static void *PINT_worker_queues_thread_function(void * ptr)
@@ -263,8 +312,7 @@ static void *PINT_worker_queues_thread_f
PINT_queue_entry_t **qentries = NULL;
int i = 0;
int ret, service_time, error;
- struct timeval now;
- struct timespec wait_for_queue_timeout, wait_interval;
+ struct timespec wait_interval;
wait_interval.tv_sec = (WAIT_FOR_QUEUE_INTERVAL / 1e6);
wait_interval.tv_nsec =
Index: pint-worker.h
===================================================================
RCS file: /projects/cvsroot/pvfs2/src/common/mgmt/Attic/pint-worker.h,v
diff -p -u -r1.1.2.1 -r1.1.2.2
--- pint-worker.h 29 Jul 2008 22:48:57 -0000 1.1.2.1
+++ pint-worker.h 26 Aug 2008 14:21:46 -0000 1.1.2.2
@@ -146,8 +146,13 @@ struct PINT_worker_impl
int (*do_work) (struct PINT_manager_s *manager,
PINT_worker_inst *inst,
PINT_context_id context_id,
- PINT_op_id op_id,
+ PINT_operation_t *op,
int microsecs);
+
+ int (*cancel) (struct PINT_manager_s *manager,
+ PINT_worker_inst *inst,
+ PINT_queue_id queue_id,
+ PINT_operation_t *op);
};
#endif
More information about the Pvfs2-cvs
mailing list