← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to plugin/logging_stats/scoreboard.cc

  • Committer: Jay Pipes
  • Date: 2010-04-08 16:27:25 UTC
  • mfrom: (1405.6.10 replication-pairs)
  • mto: This revision was merged to the branch mainline in revision 1457.
  • Revision ID: jpipes@serialcoder-20100408162725-sugbgn38oxjqclq2
Merge trunk and replication-pairs with conflict resolution

Show diffs side-by-side

added added

removed removed

Lines of Context:
plugin/logging_stats/scoreboard.cc
 
1
/*
 
2
 * Copyright (c) 2010, Joseph Daly <skinny.moey@gmail.com>
 
3
 * All rights reserved.
 
4
 *
 
5
 * Redistribution and use in source and binary forms, with or without
 
6
 * modification, are permitted provided that the following conditions are met:
 
7
 *
 
8
 *   * Redistributions of source code must retain the above copyright notice,
 
9
 *     this list of conditions and the following disclaimer.
 
10
 *   * Redistributions in binary form must reproduce the above copyright notice,
 
11
 *     this list of conditions and the following disclaimer in the documentation
 
12
 *     and/or other materials provided with the distribution.
 
13
 *   * Neither the name of Joseph Daly nor the names of its contributors
 
14
 *     may be used to endorse or promote products derived from this software
 
15
 *     without specific prior written permission.
 
16
 *
 
17
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 
18
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 
19
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 
20
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
 
21
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 
22
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 
23
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 
24
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 
25
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 
26
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
 
27
 * THE POSSIBILITY OF SUCH DAMAGE.
 
28
 *
 
29
 */
 
30
 
 
31
/**
 
32
 * @details
 
33
 *
 
34
 * The scoreboard is a pre-allocated vector of vectors of ScoreBoardSlots. It
 
35
 * can be thought of as a vector of buckets where each bucket contains
 
36
 * pre-allocated ScoreBoardSlots. To determine which bucket gets used for
 
37
 * recording statistics the modulus operator is used on the session_id. This 
 
38
 * will result in a bucket to search for a unused ScoreBoardSlot. 
 
39
 * 
 
40
 * Locking  
 
41
 *   
 
42
 * Each bucket has a its own lock this allows a search of bucket 1 and bucket 2
 
43
 * to happen concurrently.  
 
44
 *
 
45
 */
 
46
 
 
47
#include "config.h"
 
48
#include "scoreboard.h"
 
49
 
 
50
#include <math.h>
 
51
 
 
52
using namespace drizzled;
 
53
using namespace std;
 
54
 
 
55
Scoreboard::Scoreboard(uint32_t in_number_sessions, uint32_t in_number_buckets)
 
56
  :
 
57
    number_sessions(in_number_sessions),
 
58
    number_buckets(in_number_buckets)
 
59
{
 
60
 
 
61
  /* calculate the number of elements in each bucket */
 
62
  uint32_t number_per_bucket= static_cast<uint32_t> ( ceil( static_cast<double>(number_sessions) / static_cast<double>(number_buckets) ) );
 
63
 
 
64
  /* populate the vector of scoreboard vectors */
 
65
  for (uint32_t j= 0; j < number_buckets; ++j)
 
66
  {
 
67
    vector<ScoreboardSlot* > *scoreboard_vector= new vector<ScoreboardSlot* >();
 
68
 
 
69
    /* preallocate the individual vectors */
 
70
    vector<ScoreboardSlot* >::iterator scoreboard_vector_iterator= scoreboard_vector->begin();
 
71
    for (uint32_t h= 0; h < number_per_bucket; ++h)
 
72
    {
 
73
      ScoreboardSlot *scoreboard_slot= new ScoreboardSlot();
 
74
      scoreboard_vector_iterator= scoreboard_vector->insert(scoreboard_vector_iterator, scoreboard_slot);
 
75
    }  
 
76
    scoreboard_vector->resize(number_per_bucket);
 
77
 
 
78
 
 
79
    /* insert the vector into the vector of scoreboard vectors */
 
80
    vector<vector<ScoreboardSlot* >* >::iterator vector_of_scoreboard_vectors_iterator= 
 
81
      vector_of_scoreboard_vectors.begin();
 
82
 
 
83
    vector_of_scoreboard_vectors_iterator= 
 
84
      vector_of_scoreboard_vectors.insert(vector_of_scoreboard_vectors_iterator, scoreboard_vector); 
 
85
  }
 
86
  vector_of_scoreboard_vectors.resize(number_buckets);
 
87
  
 
88
  /* populate the scoreboard locks vector each ScoreboardSlot vector gets a lock */
 
89
  vector<pthread_rwlock_t* >::iterator vector_of_scoreboard_locks_iterator= vector_of_scoreboard_locks.begin();
 
90
  for (uint32_t k= 0; k < number_buckets; ++k)
 
91
  {
 
92
    pthread_rwlock_t* lock= new pthread_rwlock_t();
 
93
    (void) pthread_rwlock_init(lock, NULL);
 
94
    vector_of_scoreboard_locks_iterator= 
 
95
      vector_of_scoreboard_locks.insert(vector_of_scoreboard_locks_iterator, lock);   
 
96
  } 
 
97
  vector_of_scoreboard_locks.resize(number_buckets);
 
98
}
 
99
 
 
100
Scoreboard::~Scoreboard()
 
101
{
 
102
  vector<vector<ScoreboardSlot* >* >::iterator v_of_scoreboard_v_begin_it= vector_of_scoreboard_vectors.begin();
 
103
  vector<vector<ScoreboardSlot* >* >::iterator v_of_scoreboard_v_end_it= vector_of_scoreboard_vectors.end();
 
104
 
 
105
  for (; v_of_scoreboard_v_begin_it != v_of_scoreboard_v_end_it; ++v_of_scoreboard_v_begin_it)
 
106
  {
 
107
    vector<ScoreboardSlot* > *scoreboard_vector= *v_of_scoreboard_v_begin_it; 
 
108
 
 
109
    vector<ScoreboardSlot* >::iterator scoreboard_vector_it= scoreboard_vector->begin();
 
110
    vector<ScoreboardSlot* >::iterator scoreboard_vector_end= scoreboard_vector->end();
 
111
    for (; scoreboard_vector_it != scoreboard_vector_end; ++scoreboard_vector_it)
 
112
    {
 
113
      delete *scoreboard_vector_it; 
 
114
    }
 
115
    
 
116
    scoreboard_vector->clear();
 
117
    delete scoreboard_vector;
 
118
  } // vector_of_scoreboard_vectors is not on the stack and does not deletion
 
119
  
 
120
  vector<pthread_rwlock_t* >::iterator vector_of_scoreboard_locks_it= vector_of_scoreboard_locks.begin();
 
121
  vector<pthread_rwlock_t* >::iterator vector_of_scoreboard_locks_end= vector_of_scoreboard_locks.end();
 
122
 
 
123
  for (; vector_of_scoreboard_locks_it != vector_of_scoreboard_locks_end; ++vector_of_scoreboard_locks_it)
 
124
  {
 
125
    pthread_rwlock_t* lock= *vector_of_scoreboard_locks_it;
 
126
    (void) pthread_rwlock_destroy(lock);
 
127
    delete lock;
 
128
  }
 
129
}
 
130
 
 
131
ScoreboardSlot* Scoreboard::findScoreboardSlotToLog(Session *session)
 
132
{
 
133
  /* our bucket */
 
134
  uint32_t bucket_number= session->getSessionId() % number_buckets; 
 
135
 
 
136
  /* our vector corresponding to bucket_number */
 
137
  vector<ScoreboardSlot* > *scoreboard_vector= vector_of_scoreboard_vectors.at(bucket_number); 
 
138
 
 
139
  /* out lock corresponding to bucket_number */
 
140
  pthread_rwlock_t *LOCK_scoreboard_vector= vector_of_scoreboard_locks.at(bucket_number);
 
141
 
 
142
  pthread_rwlock_wrlock(LOCK_scoreboard_vector);
 
143
  ScoreboardSlot *scoreboard_slot= NULL;
 
144
  int32_t our_slot= UNINITIALIZED;
 
145
  int32_t open_slot= UNINITIALIZED;
 
146
 
 
147
  uint32_t current_slot= 0;
 
148
  for (vector<ScoreboardSlot *>::iterator it= scoreboard_vector->begin();
 
149
       it != scoreboard_vector->end(); ++it, current_slot++)
 
150
  {
 
151
    scoreboard_slot= *it;
 
152
 
 
153
    if (scoreboard_slot->isInUse() == true)
 
154
    {
 
155
      /* Check if this session is the one using this slot */
 
156
      if (scoreboard_slot->getSessionId() == session->getSessionId())
 
157
      {
 
158
        our_slot= current_slot;
 
159
        break;
 
160
      }
 
161
      else
 
162
      {
 
163
        continue;
 
164
      }
 
165
    }
 
166
    else
 
167
    {
 
168
      /* save off the open slot */
 
169
      if (open_slot == UNINITIALIZED)
 
170
      {
 
171
        open_slot= current_slot;
 
172
      }
 
173
      continue;
 
174
    }
 
175
  }
 
176
 
 
177
  if (our_slot != UNINITIALIZED)
 
178
  {
 
179
    pthread_rwlock_unlock(LOCK_scoreboard_vector);
 
180
  }
 
181
  else if (open_slot != UNINITIALIZED)
 
182
  {
 
183
    scoreboard_slot= scoreboard_vector->at(open_slot);
 
184
    scoreboard_slot->setInUse(true);
 
185
    scoreboard_slot->setSessionId(session->getSessionId());
 
186
    scoreboard_slot->setUser(session->getSecurityContext().getUser());
 
187
    scoreboard_slot->setIp(session->getSecurityContext().getIp());
 
188
    pthread_rwlock_unlock(LOCK_scoreboard_vector);
 
189
  }
 
190
  else
 
191
  {
 
192
    pthread_rwlock_unlock(LOCK_scoreboard_vector);
 
193
    /* there was no available slot for this session */
 
194
    return NULL;
 
195
  }
 
196
 
 
197
  return scoreboard_slot;
 
198
}
 
199
 
 
200
ScoreboardSlot* Scoreboard::findAndResetScoreboardSlot(Session *session)
 
201
{
 
202
  /* our bucket */
 
203
  uint32_t bucket_number= session->getSessionId() % number_buckets;
 
204
 
 
205
  /* our vector corresponding to bucket_number */
 
206
  vector<ScoreboardSlot* > *scoreboard_vector= vector_of_scoreboard_vectors.at(bucket_number);
 
207
 
 
208
  ScoreboardSlot *scoreboard_slot;
 
209
  ScoreboardSlot *return_scoreboard_slot= NULL;
 
210
 
 
211
  for (vector<ScoreboardSlot *>::iterator it= scoreboard_vector->begin();
 
212
       it != scoreboard_vector->end(); ++it)
 
213
  {
 
214
    scoreboard_slot= *it;
 
215
 
 
216
    if (scoreboard_slot->getSessionId() == session->getSessionId())
 
217
    {
 
218
      return_scoreboard_slot = new ScoreboardSlot(*scoreboard_slot);
 
219
      scoreboard_slot->reset(); 
 
220
      break;
 
221
    }
 
222
  }
 
223
 
 
224
  return return_scoreboard_slot;
 
225
}