| Index: trunk/phase3/maintenance/locking/LockServerDaemon.php |
| — | — | @@ -1,5 +1,7 @@ |
| 2 | 2 | <?php |
| 3 | | - |
| | 3 | +/** |
| | 4 | + * This code should not require MediaWiki setup or PHP files. |
| | 5 | + */ |
| 4 | 6 | if ( php_sapi_name() !== 'cli' ) { |
| 5 | 7 | die( "This is not a valid entry point.\n" ); |
| 6 | 8 | } |
| — | — | @@ -10,13 +12,12 @@ |
| 11 | 13 | LockServerDaemon::init( |
| 12 | 14 | getopt( '', array( |
| 13 | 15 | 'address:', 'port:', 'authKey:', |
| 14 | | - 'connTimeout::', 'lockTimeout::', 'maxClients::', 'maxBacklog::', 'maxLocks::', |
| | 16 | + 'lockTimeout::', 'maxClients::', 'maxBacklog::', 'maxLocks::', |
| 15 | 17 | ) ) |
| 16 | 18 | )->main(); |
| 17 | 19 | |
| 18 | 20 | /** |
| 19 | | - * Simple lock server daemon that accepts lock/unlock requests. |
| 20 | | - * This should not require MediaWiki setup or PHP files. |
| | 21 | + * Simple lock server daemon that accepts lock/unlock requests |
| 21 | 22 | */ |
| 22 | 23 | class LockServerDaemon { |
| 23 | 24 | /** @var resource */ |
| — | — | @@ -29,10 +30,9 @@ |
| 30 | 31 | /** @var LockHolder */ |
| 31 | 32 | protected $lockHolder; |
| 32 | 33 | |
| 33 | | - protected $address; // string (IP/hostname) |
| | 34 | + protected $address; // string IP address |
| 34 | 35 | protected $port; // integer |
| 35 | 36 | protected $authKey; // string key |
| 36 | | - protected $connTimeout; // array ( 'sec' => integer, 'usec' => integer ) |
| 37 | 37 | protected $lockTimeout; // integer number of seconds |
| 38 | 38 | protected $maxBacklog; // integer |
| 39 | 39 | protected $maxClients; // integer |
| — | — | @@ -40,6 +40,7 @@ |
| 41 | 41 | protected $startTime; // integer UNIX timestamp |
| 42 | 42 | protected $ticks = 0; // integer counter |
| 43 | 43 | |
| | 44 | + /* @var LockServerDaemon */ |
| 44 | 45 | protected static $instance = null; |
| 45 | 46 | |
| 46 | 47 | /** |
| — | — | @@ -54,7 +55,7 @@ |
| 55 | 56 | if ( !isset( $config[$par] ) ) { |
| 56 | 57 | die( "Usage: php LockServerDaemon.php " . |
| 57 | 58 | "--address <address> --port <port> --authkey <key> " . |
| 58 | | - "[--connTimeout <seconds>] [--lockTimeout <seconds>] " . |
| | 59 | + "[--lockTimeout <seconds>] " . |
| 59 | 60 | "[--maxLocks <integer>] [--maxClients <integer>] [--maxBacklog <integer>]" |
| 60 | 61 | ); |
| 61 | 62 | } |
| — | — | @@ -72,13 +73,6 @@ |
| 73 | 74 | $this->port = $config['port']; |
| 74 | 75 | $this->authKey = $config['authKey']; |
| 75 | 76 | // Parameters with defaults... |
| 76 | | - $connTimeout = isset( $config['connTimeout'] ) |
| 77 | | - ? $config['connTimeout'] |
| 78 | | - : 1.5; |
| 79 | | - $this->connTimeout = array( |
| 80 | | - 'sec' => floor( $connTimeout ), |
| 81 | | - 'usec' => floor( ( $connTimeout - floor( $connTimeout ) ) * 1e6 ) |
| 82 | | - ); |
| 83 | 77 | $this->lockTimeout = isset( $config['lockTimeout'] ) |
| 84 | 78 | ? (int)$config['lockTimeout'] |
| 85 | 79 | : 60; |
| — | — | @@ -90,7 +84,7 @@ |
| 91 | 85 | : 100; |
| 92 | 86 | $maxLocks = isset( $config['maxLocks'] ) |
| 93 | 87 | ? (int)$config['maxLocks'] |
| 94 | | - : 5000; |
| | 88 | + : 10000; |
| 95 | 89 | |
| 96 | 90 | $this->lockHolder = new LockHolder( $maxLocks ); |
| 97 | 91 | } |
| — | — | @@ -98,7 +92,7 @@ |
| 99 | 93 | /** |
| 100 | 94 | * @return void |
| 101 | 95 | */ |
| 102 | | - protected function setupSocket() { |
| | 96 | + protected function setupServerSocket() { |
| 103 | 97 | if ( !function_exists( 'socket_create' ) ) { |
| 104 | 98 | throw new Exception( "PHP sockets extension missing from PHP CLI mode." ); |
| 105 | 99 | } |
| — | — | @@ -116,75 +110,69 @@ |
| 117 | 111 | socket_strerror( socket_last_error( $sock ) ) ); |
| 118 | 112 | } |
| 119 | 113 | $this->sock = $sock; |
| 120 | | - |
| 121 | 114 | $this->startTime = time(); |
| 122 | 115 | } |
| 123 | 116 | |
| 124 | 117 | /** |
| 125 | | - * @return void |
| | 118 | + * Entry-point function that listens to the server socket, accepts |
| | 119 | + * new clients, and recieves/responds to requests to lock resources. |
| 126 | 120 | */ |
| 127 | 121 | public function main() { |
| 128 | | - // Setup socket and start listing |
| 129 | | - $this->setupSocket(); |
| 130 | | - // Create a list of all the clients that will be connected to us. |
| 131 | | - $clients = array( $this->sock ); // start off with listening socket |
| | 122 | + $this->setupServerSocket(); // setup listening socket |
| | 123 | + $socketArray = new SocketArray(); // sockets being serviced |
| | 124 | + $socketArray->addSocket( $this->sock ); // add listening socket |
| 132 | 125 | do { |
| 133 | | - // Create a copy, so $clients doesn't get modified by socket_select() |
| 134 | | - $read = $clients; // clients-with-data (plus listening socket) |
| 135 | | - // Get a list of all the clients that have data to be read from |
| 136 | | - $changed = socket_select( $read, $write = NULL, $except = NULL, NULL ); |
| 137 | | - if ( $changed === false ) { |
| 138 | | - trigger_error( 'socket_listen(): ' . socket_strerror( socket_last_error() ) ); |
| 139 | | - continue; |
| 140 | | - } elseif ( $changed < 1 ) { |
| | 126 | + list( $read, $write ) = $socketArray->socketsForSelect(); |
| | 127 | + if ( socket_select( $read, $write, $except = NULL, NULL ) < 1 ) { |
| 141 | 128 | continue; // wait |
| 142 | 129 | } |
| 143 | 130 | // Check if there is a client trying to connect... |
| 144 | | - if ( in_array( $this->sock, $read ) && count( $clients ) < $this->maxClients ) { |
| 145 | | - // Accept the new client... |
| 146 | | - $newsock = socket_accept( $this->sock ); |
| 147 | | - if ( $newsock ) { |
| 148 | | - socket_set_option( $newsock, SOL_SOCKET, SO_KEEPALIVE, 1 ); |
| 149 | | - socket_set_option( $newsock, SOL_SOCKET, SO_RCVTIMEO, $this->connTimeout ); |
| 150 | | - socket_set_option( $newsock, SOL_SOCKET, SO_SNDTIMEO, $this->connTimeout ); |
| 151 | | - $clients[] = $newsock; |
| 152 | | - // Remove the listening socket from the clients-with-data array... |
| 153 | | - $key = array_search( $this->sock, $read ); |
| 154 | | - unset( $read[$key] ); |
| | 131 | + if ( in_array( $this->sock, $read ) && $socketArray->size() < $this->maxClients ) { |
| | 132 | + $newSock = socket_accept( $this->sock ); |
| | 133 | + if ( $newSock ) { |
| | 134 | + socket_set_option( $newSock, SOL_SOCKET, SO_KEEPALIVE, 1 ); |
| | 135 | + socket_set_nonblock( $newSock ); // don't block on read()/write() |
| | 136 | + $socketArray->addSocket( $newSock ); |
| 155 | 137 | } |
| 156 | 138 | } |
| 157 | 139 | // Loop through all the clients that have data to read... |
| 158 | 140 | foreach ( $read as $read_sock ) { |
| 159 | | - // Read until newline or 65535 bytes are recieved. |
| 160 | | - // socket_read show errors when the client is disconnected. |
| 161 | | - $data = @socket_read( $read_sock, 65535, PHP_NORMAL_READ ); |
| | 141 | + if ( $read_sock === $this->sock ) { |
| | 142 | + continue; // skip listening socket |
| | 143 | + } |
| | 144 | + // Avoids PHP_NORMAL_READ per https://bugs.php.net/bug.php?id=33471 |
| | 145 | + $data = socket_read( $read_sock, 65535 ); |
| 162 | 146 | // Check if the client is disconnected |
| 163 | | - if ( $data === false ) { |
| 164 | | - // Remove client from $clients list |
| 165 | | - $key = array_search( $read_sock, $clients ); |
| 166 | | - unset( $clients[$key] ); |
| 167 | | - // Remove socket's session from tracking (if it exists) |
| 168 | | - $session = array_search( $read_sock, $this->sessions ); |
| 169 | | - if ( $session !== false ) { |
| 170 | | - unset( $this->sessions[$session] ); |
| 171 | | - // Record recently killed sessions that still have locks |
| 172 | | - if ( isset( $this->sessionIndexSh[$session] ) |
| 173 | | - || isset( $this->sessionIndexEx[$session] ) ) |
| 174 | | - { |
| 175 | | - $this->deadSessions[$session] = time(); |
| 176 | | - } |
| 177 | | - } |
| 178 | | - } else { |
| | 147 | + if ( $data === false || $data === '' ) { |
| | 148 | + $socketArray->closeSocket( $read_sock ); |
| | 149 | + $this->recordDeadSocket( $read_sock ); // remove session |
| | 150 | + // Check if we reached the end of a message |
| | 151 | + } elseif ( substr( $data, -1 ) === "\n" ) { |
| | 152 | + // Newline is the last char (given ping-pong message usage) |
| | 153 | + $cmd = $socketArray->readRcvBuffer( $read_sock ) . $data; |
| 179 | 154 | // Perform the requested command... |
| 180 | | - $response = $this->doCommand( trim( $data ), $read_sock ); |
| | 155 | + $response = $this->doCommand( rtrim( $cmd ), $read_sock ); |
| 181 | 156 | // Send the response to the client... |
| 182 | | - if ( socket_write( $read_sock, "$response\n" ) === false ) { |
| 183 | | - trigger_error( 'socket_write(): ' . |
| 184 | | - socket_strerror( socket_last_error( $read_sock ) ) ); |
| 185 | | - } |
| | 157 | + $socketArray->appendSndBuffer( $read_sock, $response . "\n" ); |
| | 158 | + // Otherwise, we just have more message data to append |
| | 159 | + } elseif ( !$socketArray->appendRcvBuffer( $read_sock, $data ) ) { |
| | 160 | + $socketArray->closeSocket( $read_sock ); // too big |
| | 161 | + $this->recordDeadSocket( $read_sock ); // remove session |
| 186 | 162 | } |
| 187 | 163 | } |
| 188 | | - // Prune dead locks every 10 socket events... |
| | 164 | + // Loop through all the clients that have data to write... |
| | 165 | + foreach ( $write as $write_sock ) { |
| | 166 | + $bytes = socket_write( $write_sock, $socketArray->readSndBuffer( $write_sock ) ); |
| | 167 | + // Check if the client is disconnected |
| | 168 | + if ( $bytes === false ) { |
| | 169 | + $socketArray->closeSocket( $write_sock ); |
| | 170 | + $this->recordDeadSocket( $write_sock ); // remove session |
| | 171 | + // Otherwise, truncate these bytes from the start of the write buffer |
| | 172 | + } else { |
| | 173 | + $socketArray->consumeSndBuffer( $write_sock, $bytes ); |
| | 174 | + } |
| | 175 | + } |
| | 176 | + // Prune dead locks every few socket events... |
| 189 | 177 | if ( ++$this->ticks >= 9 ) { |
| 190 | 178 | $this->ticks = 0; |
| 191 | 179 | $this->purgeExpiredLocks(); |
| — | — | @@ -206,6 +194,7 @@ |
| 207 | 195 | // On first command, track the session => sock correspondence |
| 208 | 196 | if ( !isset( $this->sessions[$session] ) ) { |
| 209 | 197 | $this->sessions[$session] = $sourceSock; |
| | 198 | + unset( $this->deadSessions[$session] ); // renew if dead |
| 210 | 199 | } |
| 211 | 200 | if ( $function === 'ACQUIRE' ) { |
| 212 | 201 | return $this->lockHolder->lock( $session, $type, $resources ); |
| — | — | @@ -260,13 +249,33 @@ |
| 261 | 250 | } |
| 262 | 251 | |
| 263 | 252 | /** |
| | 253 | + * Remove a socket's corresponding session from tracking and |
| | 254 | + * store it in the dead session tracking if it still has locks. |
| | 255 | + * |
| | 256 | + * @param $socket resource |
| | 257 | + * @return book |
| | 258 | + */ |
| | 259 | + protected function recordDeadSocket( $socket ) { |
| | 260 | + $session = array_search( $socket, $this->sessions ); |
| | 261 | + if ( $session !== false ) { |
| | 262 | + unset( $this->sessions[$session] ); |
| | 263 | + // Record recently killed sessions that still have locks |
| | 264 | + if ( $this->lockHolder->sessionHasLocks( $session ) ) { |
| | 265 | + $this->deadSessions[$session] = time(); |
| | 266 | + } |
| | 267 | + return true; |
| | 268 | + } |
| | 269 | + return false; |
| | 270 | + } |
| | 271 | + |
| | 272 | + /** |
| 264 | 273 | * Clear locks for sessions that have been dead for a while |
| 265 | 274 | * |
| 266 | 275 | * @return integer Number of sessions purged |
| 267 | 276 | */ |
| 268 | 277 | protected function purgeExpiredLocks() { |
| | 278 | + $count = 0; |
| 269 | 279 | $now = time(); |
| 270 | | - $count = 0; |
| 271 | 280 | foreach ( $this->deadSessions as $session => $timestamp ) { |
| 272 | 281 | if ( ( $now - $timestamp ) > $this->lockTimeout ) { |
| 273 | 282 | $this->lockHolder->release( $session ); |
| — | — | @@ -288,9 +297,144 @@ |
| 289 | 298 | } |
| 290 | 299 | |
| 291 | 300 | /** |
| 292 | | - * LockServerDaemon helper class that keeps track of the locks. |
| 293 | | - * This should not require MediaWiki setup or PHP files. |
| | 301 | + * LockServerDaemon helper class that keeps track socket states |
| 294 | 302 | */ |
| | 303 | +class SocketArray { |
| | 304 | + /* @var Array */ |
| | 305 | + protected $clients = array(); // array of client sockets |
| | 306 | + /* @var Array */ |
| | 307 | + protected $rBuffers = array(); // corresponding socket read buffers |
| | 308 | + /* @var Array */ |
| | 309 | + protected $wBuffers = array(); // corresponding socket write buffers |
| | 310 | + |
| | 311 | + const BUFFER_SIZE = 65535; |
| | 312 | + |
| | 313 | + /** |
| | 314 | + * @return Array (list of sockets to read, list of sockets to write) |
| | 315 | + */ |
| | 316 | + public function socketsForSelect() { |
| | 317 | + $rSockets = array(); |
| | 318 | + $wSockets = array(); |
| | 319 | + foreach ( $this->clients as $key => $socket ) { |
| | 320 | + if ( $this->wBuffers[$key] !== '' ) { |
| | 321 | + $wSockets[] = $socket; // wait for writing to unblock |
| | 322 | + } else { |
| | 323 | + $rSockets[] = $socket; // wait for reading to unblock |
| | 324 | + } |
| | 325 | + } |
| | 326 | + return array( $rSockets, $wSockets ); |
| | 327 | + } |
| | 328 | + |
| | 329 | + /** |
| | 330 | + * @return integer Number of client sockets |
| | 331 | + */ |
| | 332 | + public function size() { |
| | 333 | + return count( $this->clients ); |
| | 334 | + } |
| | 335 | + |
| | 336 | + /** |
| | 337 | + * @param $sock resource |
| | 338 | + * @return bool |
| | 339 | + */ |
| | 340 | + public function addSocket( $sock ) { |
| | 341 | + $this->clients[] = $sock; |
| | 342 | + $this->rBuffers[] = ''; |
| | 343 | + $this->wBuffers[] = ''; |
| | 344 | + return true; |
| | 345 | + } |
| | 346 | + |
| | 347 | + /** |
| | 348 | + * @param $sock resource |
| | 349 | + * @return bool |
| | 350 | + */ |
| | 351 | + public function closeSocket( $sock ) { |
| | 352 | + $key = array_search( $sock, $this->clients ); |
| | 353 | + if ( $key === false ) { |
| | 354 | + return false; |
| | 355 | + } |
| | 356 | + socket_close( $sock ); |
| | 357 | + unset( $this->clients[$key] ); |
| | 358 | + unset( $this->rBuffers[$key] ); |
| | 359 | + unset( $this->wBuffers[$key] ); |
| | 360 | + return true; |
| | 361 | + } |
| | 362 | + |
| | 363 | + /** |
| | 364 | + * @param $sock resource |
| | 365 | + * @param $data string |
| | 366 | + * @return bool |
| | 367 | + */ |
| | 368 | + public function appendRcvBuffer( $sock, $data ) { |
| | 369 | + $key = array_search( $sock, $this->clients ); |
| | 370 | + if ( $key === false ) { |
| | 371 | + return false; |
| | 372 | + } elseif ( ( strlen( $this->rBuffers[$key] ) + strlen( $data ) ) > self::BUFFER_SIZE ) { |
| | 373 | + return false; |
| | 374 | + } |
| | 375 | + $this->rBuffers[$key] .= $data; |
| | 376 | + return true; |
| | 377 | + } |
| | 378 | + |
| | 379 | + /** |
| | 380 | + * @param $sock resource |
| | 381 | + * @return string|false |
| | 382 | + */ |
| | 383 | + public function readRcvBuffer( $sock ) { |
| | 384 | + $key = array_search( $sock, $this->clients ); |
| | 385 | + if ( $key === false ) { |
| | 386 | + return false; |
| | 387 | + } |
| | 388 | + $data = $this->rBuffers[$key]; |
| | 389 | + $this->rBuffers[$key] = ''; // consume data |
| | 390 | + return $data; |
| | 391 | + } |
| | 392 | + |
| | 393 | + /** |
| | 394 | + * @param $sock resource |
| | 395 | + * @param $data string |
| | 396 | + * @return bool |
| | 397 | + */ |
| | 398 | + public function appendSndBuffer( $sock, $data ) { |
| | 399 | + $key = array_search( $sock, $this->clients ); |
| | 400 | + if ( $key === false ) { |
| | 401 | + return false; |
| | 402 | + } elseif ( ( strlen( $this->wBuffers[$key] ) + strlen( $data ) ) > self::BUFFER_SIZE ) { |
| | 403 | + return false; |
| | 404 | + } |
| | 405 | + $this->wBuffers[$key] .= $data; |
| | 406 | + return true; |
| | 407 | + } |
| | 408 | + |
| | 409 | + /** |
| | 410 | + * @param $sock resource |
| | 411 | + * @return bool |
| | 412 | + */ |
| | 413 | + public function readSndBuffer( $sock ) { |
| | 414 | + $key = array_search( $sock, $this->clients ); |
| | 415 | + if ( $key === false ) { |
| | 416 | + return false; |
| | 417 | + } |
| | 418 | + return $this->wBuffers[$key]; |
| | 419 | + } |
| | 420 | + |
| | 421 | + /** |
| | 422 | + * @param $sock resource |
| | 423 | + * @param $bytes integer |
| | 424 | + * @return bool |
| | 425 | + */ |
| | 426 | + public function consumeSndBuffer( $sock, $bytes ) { |
| | 427 | + $key = array_search( $sock, $this->clients ); |
| | 428 | + if ( $key === false ) { |
| | 429 | + return false; |
| | 430 | + } |
| | 431 | + $this->wBuffers[$key] = (string)substr( $this->wBuffers[$key], $bytes ); |
| | 432 | + return true; |
| | 433 | + } |
| | 434 | +} |
| | 435 | + |
| | 436 | +/** |
| | 437 | + * LockServerDaemon helper class that keeps track of the locks |
| | 438 | + */ |
| 295 | 439 | class LockHolder { |
| 296 | 440 | /** @var Array */ |
| 297 | 441 | protected $shLocks = array(); // (key => session => 1) |
| — | — | @@ -314,6 +458,15 @@ |
| 315 | 459 | |
| 316 | 460 | /** |
| 317 | 461 | * @param $session string |
| | 462 | + * @return bool |
| | 463 | + */ |
| | 464 | + public function sessionHasLocks( $session ) { |
| | 465 | + return isset( $this->sessionIndexSh[$session] ) |
| | 466 | + || isset( $this->sessionIndexEx[$session] ); |
| | 467 | + } |
| | 468 | + |
| | 469 | + /** |
| | 470 | + * @param $session string |
| 318 | 471 | * @param $type string |
| 319 | 472 | * @param $keys Array |
| 320 | 473 | * @return string |