lustre/utils/lctl_thread.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * This file is part of Lustre, http://www.lustre.org/
  24  *
  25  * lustre/utils/lctl_thread.c
  26  *
  27  * Author: Rajeev Mishra <rajeevm@hpe.com>
  28  */
  29  #include <errno.h>
  30  #include <stdio.h>
  31  #include <stdarg.h>
  32  #include <ctype.h>
  33  #include "lctl_thread.h"
  34  #include <stdlib.h>
  35  #include <libcfs/util/string.h>
  36 #if HAVE_LIBPTHREAD
  37 /**
  38  * Initialize the given set_param work queue.
  39  *
  40  * \param[out] wq  the work queue to initialize
  41  * \param[in] popt the options passed to set_param
  42  *
  43  * \retval 0 if successful
  44  * \retval -errno if unsuccessful
  45  */
  46 int spwq_init(struct sp_workq *wq, struct param_opts *popt)
  47 {
  48         if (!wq)
  49                 return -EINVAL;
  50
  51         memset(wq, 0, sizeof(*wq));
  52         wq->spwq_popt = popt;
  53
  54         /* pthread_mutex_init returns 0 for success, or errno for failure */
  55         return -pthread_mutex_init(&wq->spwq_mutex, NULL);
  56 }
  57
  58 /**
  59  * Destroy and free space used by a set_param work queue.
  60  *
  61  * \param[in] wq the work queue to destroy
  62  *
  63  * \retval 0 if successful
  64  * \retval -errno if unsuccessful
  65  */
  66 int spwq_destroy(struct sp_workq *wq)
  67 {
  68         int rc;
  69
  70         if (!wq)
  71                 return 0;
  72
  73         if (wq->spwq_items) {
  74                 int i;
  75
  76                 for (i = 0; i < wq->spwq_len; i++) {
  77                         free(wq->spwq_items[i].spwi_path);
  78                         free(wq->spwq_items[i].spwi_param_name);
  79                         /* wq->spwq_items[i].spwi_value was not malloc'd */
  80                 }
  81                 free(wq->spwq_items);
  82         }
  83
  84         /* pthread_mutex_destroy returns 0 for success, or errno for failure */
  85         rc = -pthread_mutex_destroy(&wq->spwq_mutex);
  86
  87         memset(wq, 0, sizeof(*wq));
  88
  89         return rc;
  90 }
  91
  92 /**
  93  * Expand the size of a work queue to fit the requested number of items.
  94  *
  95  * \param[in,out] wq    the work queue to expand
  96  * \param[in] num_items the number of items to make room for in \a wq
  97  *
  98  * \retval 0 if successful
  99  * \retval -errno if unsuccessful
 100  */
 101 int spwq_expand(struct sp_workq *wq, size_t num_items)
 102 {
 103         int space;
 104         int new_size;
 105         struct sp_work_item *tmp;
 106
 107         if (!wq)
 108                 return -EINVAL;
 109
 110         space = wq->spwq_size - wq->spwq_len;
 111
 112         /* First check if there's already enough room. */
 113         if (space >= num_items)
 114                 return 0;
 115
 116         new_size = wq->spwq_len + num_items;
 117
 118         /* When spwq_items is NULL, realloc behaves like malloc */
 119         tmp = realloc(wq->spwq_items, new_size * sizeof(struct sp_work_item));
 120
 121         if (!tmp)
 122                 return -ENOMEM;
 123
 124         wq->spwq_items = tmp;
 125         wq->spwq_size = new_size;
 126
 127         return 0;
 128 }
 129
 130 /**
 131  * Add an item to a set_param work queue. Not thread-safe.
 132  *
 133  * \param[in,out] wq     the work queue to which the item should be added
 134  * \param[in] path       the full path to the parameter file (will be copied)
 135  * \param[in] param_name the name of the parameter (will be copied)
 136  * \param[in] value      the value for the parameter (will not be copied)
 137  *
 138  * \retval 0 if successful
 139  * \retval -errno if unsuccessful
 140  */
 141 int spwq_add_item(struct sp_workq *wq, char *path,
 142                          char *param_name, char *value)
 143 {
 144         char *path_copy;
 145         char *param_name_copy;
 146         int rc;
 147
 148         if (!(wq && path && param_name && value))
 149                 return -EINVAL;
 150
 151         /* Hopefully the caller has expanded the work queue before calling this
 152          * function, but make sure there's room just in case.
 153          */
 154         rc = spwq_expand(wq, 1);
 155         if (rc < 0)
 156                 return rc;
 157
 158         path_copy = strdup(path);
 159         if (!path_copy)
 160                 return -ENOMEM;
 161
 162         param_name_copy = strdup(param_name);
 163         if (!param_name_copy) {
 164                 free(path_copy);
 165                 return -ENOMEM;
 166         }
 167
 168         wq->spwq_items[wq->spwq_len].spwi_param_name = param_name_copy;
 169         wq->spwq_items[wq->spwq_len].spwi_path = path_copy;
 170         wq->spwq_items[wq->spwq_len].spwi_value = value;
 171
 172         wq->spwq_len++;
 173
 174         return 0;
 175 }
 176
 177 /**
 178  * Gets the next item from the set_param \a wq in a thread-safe manner.
 179  *
 180  * \param[in] wq  the workq from which to obtain the next item
 181  * \param[out] wi the next work item in \a wa, will be set to NULL if \wq empty
 182  *
 183  * \retval 0 if successful (empty work queue is considered successful)
 184  * \retval -errno if unsuccessful
 185  */
 186 static int spwq_next_item(struct sp_workq *wq, struct sp_work_item **wi)
 187 {
 188         int rc_lock;
 189         int rc_unlock;
 190
 191         if (!(wq && wi))
 192                 return -EINVAL;
 193
 194         *wi = NULL;
 195
 196         rc_lock = pthread_mutex_lock(&wq->spwq_mutex);
 197         if (rc_lock == 0) {
 198                 if (wq->spwq_cur_index < wq->spwq_len)
 199                         *wi = &wq->spwq_items[wq->spwq_cur_index++];
 200                 rc_unlock = pthread_mutex_unlock(&wq->spwq_mutex);
 201         }
 202
 203         return rc_lock != 0 ? -rc_lock : -rc_unlock;
 204 }
 205
 206 /**
 207  * A set_param worker thread which sets params from the workq.
 208  *
 209  * \param[in] arg a pointer to a struct sp_workq
 210  *
 211  * \retval 0 if successful
 212  * \retval -errno if unsuccessful
 213  */
 214 static void *sp_thread(void *arg)
 215 {
 216         struct sp_workq *wq = (struct sp_workq *)arg;
 217         struct param_opts *popt = wq->spwq_popt;
 218         struct sp_work_item *work_item;
 219         long int rc = 0;
 220
 221         rc = spwq_next_item(wq, &work_item);
 222         if (rc < 0)
 223                 return (void *)rc;
 224
 225         while (work_item) {
 226                 char *path = work_item->spwi_path;
 227                 char *param_name = work_item->spwi_param_name;
 228                 char *value = work_item->spwi_value;
 229                 int rc2;
 230
 231                 rc2 = write_param(path, param_name, popt, value);
 232                 if (rc2 < 0)
 233                         rc = rc2;
 234                 rc2 = spwq_next_item(wq, &work_item);
 235                 if (rc2 < 0)
 236                         rc = rc2;
 237         }
 238
 239         return (void *)rc;
 240 }
 241
 242 /**
 243  * Spawn threads and set parameters in a work queue in parallel.
 244  *
 245  * \param[in] wq the work queue containing parameters to set
 246  *
 247  * \retval 0 if successful
 248  * \retval -errno if unsuccessful
 249  */
 250 int sp_run_threads(struct sp_workq *wq)
 251 {
 252         int rc = 0;
 253         int i;
 254         int j;
 255         int num_threads;
 256         pthread_t *sp_threads;
 257
 258         if (!wq)
 259                 return -EINVAL;
 260
 261         if (wq->spwq_len == 0)
 262                 return 0;
 263
 264         num_threads = wq->spwq_popt->po_parallel_threads;
 265         if (num_threads > wq->spwq_len)
 266                 num_threads = wq->spwq_len;
 267
 268         sp_threads = malloc(sizeof(pthread_t) * num_threads);
 269         if (!sp_threads)
 270                 return -ENOMEM;
 271
 272         for (i = 0; i < num_threads; i++) {
 273                 rc = -pthread_create(&sp_threads[i], NULL, &sp_thread, wq);
 274                 if (rc != 0)
 275                         break;
 276         }
 277
 278         /* check if we failed to create any threads at all */
 279         if (i == 0)
 280                 goto out_free;
 281
 282         /* ignore thread creation errors if at least one was created */
 283         rc = 0;
 284
 285         for (j = 0; j < i; j++) {
 286                 int join_rc;
 287                 void *res = NULL;
 288
 289                 join_rc = -pthread_join(sp_threads[j], &res);
 290                 if (join_rc && rc == 0)
 291                         rc = join_rc;
 292                 if (res)
 293                         /* this error takes priority over join errors */
 294                         rc = (long int)res;
 295         }
 296
 297 out_free:
 298         free(sp_threads);
 299         return rc;
 300 }
 301 #else
 302 #define popt_is_parallel(popt) 0
 303 #define spwq_init(wq, popt) 0
 304 #define spwq_expand(wq, num_items) 0
 305 #define spwq_add_item(wq, path, param_name, value) 0
 306 #define sp_run_threads(wq) 0
 307 #define spwq_destroy(wq) 0
 308 struct sp_workq { int unused; }
 309 #endif /* HAVE_LIBPTHREAD */