← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to plugin/multi_thread/multi_thread.cc

  • Committer: LinuxJedi
  • Date: 2010-08-28 09:23:52 UTC
  • mto: (1738.1.1 build)
  • mto: This revision was merged to the branch mainline in revision 1739.
  • Revision ID: linuxjedi@linuxjedi-laptop-20100828092352-oe3zbtdy05kq9dtq
Make exit happen in main thread rather than signal handler thread thus avoiding a segfault due to a double kill of the signal handler thread

Show diffs side-by-side

added added

removed removed

Lines of Context:
plugin/multi_thread/multi_thread.cc
19
19
#include <boost/program_options.hpp>
20
20
#include <drizzled/module/option_map.h>
21
21
#include <drizzled/errmsg_print.h>
22
 
#include "drizzled/session.h"
23
 
#include "drizzled/session/cache.h"
24
 
 
25
 
#include <boost/thread.hpp>
26
 
#include <boost/bind.hpp>
27
22
 
28
23
namespace po= boost::program_options;
29
24
using namespace std;
30
25
using namespace drizzled;
31
26
 
32
27
/* Configuration variables. */
33
 
typedef constrained_check<uint32_t, 4096, 1> max_threads_constraint;
34
 
static max_threads_constraint max_threads;
 
28
static uint32_t max_threads;
 
29
 
 
30
/* Global's (TBR) */
 
31
static MultiThreadScheduler *scheduler= NULL;
35
32
 
36
33
namespace drizzled
37
34
{
38
35
  extern size_t my_thread_stack_size;
39
36
}
40
37
 
41
 
namespace multi_thread {
42
 
 
43
 
void MultiThreadScheduler::runSession(drizzled::session_id_t id)
44
 
{
45
 
  char stack_dummy;
46
 
  boost::this_thread::disable_interruption disable_by_default;
47
 
  Session::shared_ptr session(session::Cache::singleton().find(id));
48
 
 
49
 
  if (not session)
50
 
  {
51
 
    std::cerr << "Session killed before thread could execute\n";
52
 
    return;
53
 
  }
54
 
  session->pushInterrupt(&disable_by_default);
55
 
 
56
 
  if (drizzled::internal::my_thread_init())
57
 
  {
58
 
    session->disconnect(drizzled::ER_OUT_OF_RESOURCES, true);
59
 
    session->status_var.aborted_connects++;
60
 
    killSessionNow(session);
61
 
  }
62
 
  boost::this_thread::at_thread_exit(&internal::my_thread_end);
63
 
 
64
 
  session->thread_stack= (char*) &stack_dummy;
65
 
  session->run();
66
 
  killSessionNow(session);
67
 
  // @todo remove hard spin by disconnection the session first from the
68
 
  // thread.
69
 
  while (not session.unique()) {}
70
 
}
71
 
 
72
 
void MultiThreadScheduler::setStackSize()
73
 
{
74
 
  pthread_attr_t attr;
75
 
 
76
 
  (void) pthread_attr_init(&attr);
77
 
 
78
 
  /* Get the thread stack size that the OS will use and make sure
79
 
    that we update our global variable. */
80
 
  int err= pthread_attr_getstacksize(&attr, &my_thread_stack_size);
81
 
  pthread_attr_destroy(&attr);
82
 
 
83
 
  if (err != 0)
84
 
  {
85
 
    errmsg_printf(ERRMSG_LVL_ERROR, _("Unable to get thread stack size\n"));
86
 
    my_thread_stack_size= 524288; // At the time of the writing of this code, this was OSX's
87
 
  }
88
 
 
89
 
  if (my_thread_stack_size == 0)
90
 
  {
91
 
    my_thread_stack_size= 524288; // At the time of the writing of this code, this was OSX's
92
 
  }
 
38
/**
 
39
 * Function to be run as a thread for each session.
 
40
 */
 
41
namespace
 
42
{
 
43
  extern "C" pthread_handler_t session_thread(void *arg);
 
44
}
 
45
 
 
46
namespace
 
47
{
 
48
  extern "C" pthread_handler_t session_thread(void *arg)
 
49
  {
 
50
    Session *session= static_cast<Session*>(arg);
 
51
    MultiThreadScheduler *sched= static_cast<MultiThreadScheduler*>(session->scheduler);
 
52
    sched->runSession(session);
 
53
    return NULL;
 
54
  }
 
55
}
 
56
 
 
57
 
 
58
bool MultiThreadScheduler::addSession(Session *session)
 
59
{
 
60
  if (thread_count >= max_threads)
 
61
    return true;
 
62
 
93
63
#ifdef __sun
94
64
  /*
95
65
   * Solaris will return zero for the stack size in a call to
101
71
   * will be used.
102
72
   */
103
73
  if (my_thread_stack_size == 0)
104
 
  {
105
74
    my_thread_stack_size= 2 * 1024 * 1024;
106
 
  }
107
75
#endif
108
 
}
109
 
 
110
 
bool MultiThreadScheduler::addSession(Session::shared_ptr &session)
111
 
{
112
 
  if (thread_count >= max_threads)
113
 
    return true;
 
76
 
 
77
  /* Thread stack size of zero means just use the OS default */
 
78
  if (my_thread_stack_size != 0)
 
79
  {
 
80
    int err= pthread_attr_setstacksize(&attr, my_thread_stack_size);
 
81
 
 
82
    if (err != 0)
 
83
    {
 
84
      errmsg_printf(ERRMSG_LVL_ERROR,
 
85
                    _("Unable to set thread stack size to %" PRId64 "\n"),
 
86
                    static_cast<uint64_t>(my_thread_stack_size));
 
87
      return true;
 
88
    }
 
89
  }
 
90
  else
 
91
  {
 
92
    /* Get the thread stack size that the OS will use and make sure
 
93
       that we update our global variable. */
 
94
    int err= pthread_attr_getstacksize(&attr, &my_thread_stack_size);
 
95
 
 
96
    if (err != 0)
 
97
    {
 
98
      errmsg_printf(ERRMSG_LVL_ERROR, _("Unable to get thread stack size\n"));
 
99
      return true;
 
100
    }
 
101
  }
114
102
 
115
103
  thread_count.increment();
116
104
 
117
 
  session->getThread().reset(new boost::thread((boost::bind(&MultiThreadScheduler::runSession, this, session->getSessionId()))));
118
 
 
119
 
  if (not session->getThread())
120
 
  {
121
 
    thread_count.decrement();
122
 
    return true;
123
 
  }
124
 
 
125
 
  if (not session->getThread()->joinable())
 
105
  if (pthread_create(&session->real_id, &attr, session_thread,
 
106
                     static_cast<void*>(session)))
126
107
  {
127
108
    thread_count.decrement();
128
109
    return true;
132
113
}
133
114
 
134
115
 
135
 
void MultiThreadScheduler::killSession(Session *session)
136
 
{
137
 
  boost_thread_shared_ptr thread(session->getThread());
138
 
 
139
 
  if (thread)
140
 
  {
141
 
    thread->interrupt();
142
 
  }
143
 
}
144
 
 
145
 
void MultiThreadScheduler::killSessionNow(Session::shared_ptr &session)
146
 
{
147
 
  killSession(session.get());
 
116
void MultiThreadScheduler::killSessionNow(Session *session)
 
117
{
148
118
  /* Locks LOCK_thread_count and deletes session */
149
119
  Session::unlink(session);
150
120
  thread_count.decrement();
 
121
  internal::my_thread_end();
 
122
  pthread_exit(0);
 
123
  /* We should never reach this point. */
151
124
}
152
125
 
153
126
MultiThreadScheduler::~MultiThreadScheduler()
154
127
{
155
 
  boost::mutex::scoped_lock scopedLock(drizzled::session::Cache::singleton().mutex());
 
128
  LOCK_thread_count.lock();
156
129
  while (thread_count)
157
130
  {
158
 
    COND_thread_count.wait(scopedLock);
 
131
    pthread_cond_wait(COND_thread_count.native_handle(), LOCK_thread_count.native_handle());
159
132
  }
 
133
 
 
134
  LOCK_thread_count.unlock();
 
135
  (void) pthread_attr_destroy(&attr);
160
136
}
161
137
 
162
 
} // multi_thread namespace
163
 
 
164
138
  
165
139
static int init(drizzled::module::Context &context)
166
140
{
167
141
  
168
 
  context.add(new multi_thread::MultiThreadScheduler("multi_thread"));
 
142
  const module::option_map &vm= context.getOptions();
 
143
  if (vm.count("max-threads"))
 
144
  {
 
145
    if (max_threads > 4096 || max_threads < 1)
 
146
    {
 
147
      errmsg_printf(ERRMSG_LVL_ERROR, _("Invalid value for max-threads\n"));
 
148
      exit(-1);
 
149
    }
 
150
  }
 
151
 
 
152
  scheduler= new MultiThreadScheduler("multi_thread");
 
153
  context.add(scheduler);
169
154
 
170
155
  return 0;
171
156
}
172
157
 
 
158
static DRIZZLE_SYSVAR_UINT(max_threads, max_threads,
 
159
                           PLUGIN_VAR_RQCMDARG,
 
160
                           N_("Maximum number of user threads available."),
 
161
                           NULL, NULL, 2048, 1, 4096, 0);
 
162
 
173
163
static void init_options(drizzled::module::option_context &context)
174
164
{
175
165
  context("max-threads",
176
 
          po::value<max_threads_constraint>(&max_threads)->default_value(2048),
 
166
          po::value<uint32_t>(&max_threads)->default_value(2048),
177
167
          N_("Maximum number of user threads available."));
178
168
}
179
169
 
 
170
static drizzle_sys_var* sys_variables[]= {
 
171
  DRIZZLE_SYSVAR(max_threads),
 
172
  NULL
 
173
};
 
174
 
180
175
DRIZZLE_DECLARE_PLUGIN
181
176
{
182
177
  DRIZZLE_VERSION_ID,
186
181
  "One Thread Per Session Scheduler",
187
182
  PLUGIN_LICENSE_GPL,
188
183
  init, /* Plugin Init */
189
 
  NULL,   /* system variables */
 
184
  sys_variables,   /* system variables */
190
185
  init_options    /* config options */
191
186
}
192
187
DRIZZLE_DECLARE_PLUGIN_END;