Automattic /
jetpack
These results are based on our legacy PHP analysis, consider migrating to our new PHP analysis engine instead. Learn more
| 1 | <?php |
||
| 2 | |||
| 3 | /** |
||
| 4 | * A buffer of items from the queue that can be checked out |
||
| 5 | */ |
||
| 6 | class Jetpack_Sync_Queue_Buffer { |
||
| 7 | public $id; |
||
| 8 | public $items_with_ids; |
||
| 9 | |||
| 10 | public function __construct( $id, $items_with_ids ) { |
||
| 11 | $this->id = $id; |
||
| 12 | $this->items_with_ids = $items_with_ids; |
||
| 13 | } |
||
| 14 | |||
| 15 | public function get_items() { |
||
| 16 | return array_combine( $this->get_item_ids(), $this->get_item_values() ); |
||
| 17 | } |
||
| 18 | |||
| 19 | public function get_item_values() { |
||
| 20 | return Jetpack_Sync_Utils::get_item_values( $this->items_with_ids ); |
||
| 21 | } |
||
| 22 | |||
| 23 | public function get_item_ids() { |
||
| 24 | return Jetpack_Sync_Utils::get_item_ids( $this->items_with_ids ); |
||
| 25 | } |
||
| 26 | } |
||
| 27 | |||
| 28 | /** |
||
| 29 | * A persistent queue that can be flushed in increments of N items, |
||
| 30 | * and which blocks reads until checked-out buffers are checked in or |
||
| 31 | * closed. This uses raw SQL for two reasons: speed, and not triggering |
||
| 32 | * tons of added_option callbacks. |
||
| 33 | */ |
||
| 34 | class Jetpack_Sync_Queue { |
||
|
0 ignored issues
–
show
|
|||
| 35 | public $id; |
||
| 36 | private $row_iterator; |
||
| 37 | |||
| 38 | function __construct( $id ) { |
||
| 39 | $this->id = str_replace( '-', '_', $id ); // necessary to ensure we don't have ID collisions in the SQL |
||
| 40 | $this->row_iterator = 0; |
||
| 41 | } |
||
| 42 | |||
| 43 | function add( $item ) { |
||
| 44 | global $wpdb; |
||
| 45 | $added = false; |
||
| 46 | // this basically tries to add the option until enough time has elapsed that |
||
| 47 | // it has a unique (microtime-based) option key |
||
| 48 | while ( ! $added ) { |
||
| 49 | $rows_added = $wpdb->query( $wpdb->prepare( |
||
| 50 | "INSERT INTO $wpdb->options (option_name, option_value, autoload) VALUES (%s, %s,%s)", |
||
| 51 | $this->get_next_data_row_option_name(), |
||
| 52 | serialize( $item ), |
||
| 53 | 'no' |
||
| 54 | ) ); |
||
| 55 | $added = ( $rows_added !== 0 ); |
||
| 56 | } |
||
| 57 | } |
||
| 58 | |||
| 59 | // Attempts to insert all the items in a single SQL query. May be subject to query size limits! |
||
| 60 | function add_all( $items ) { |
||
| 61 | global $wpdb; |
||
| 62 | $base_option_name = $this->get_next_data_row_option_name(); |
||
| 63 | |||
| 64 | $query = "INSERT INTO $wpdb->options (option_name, option_value, autoload) VALUES "; |
||
| 65 | |||
| 66 | $rows = array(); |
||
| 67 | |||
| 68 | for ( $i = 0; $i < count( $items ); $i += 1 ) { |
||
|
0 ignored issues
–
show
It seems like you are calling the size function
count() as part of the test condition. You might want to compute the size beforehand, and not on each iteration.
If the size of the collection does not change during the iteration, it is generally a good practice to compute it beforehand, and not on each iteration: for ($i=0; $i<count($array); $i++) { // calls count() on each iteration
}
// Better
for ($i=0, $c=count($array); $i<$c; $i++) { // calls count() just once
}
Loading history...
|
|||
| 69 | $option_name = esc_sql( $base_option_name . '-' . $i ); |
||
| 70 | $option_value = esc_sql( serialize( $items[ $i ] ) ); |
||
| 71 | $rows[] = "('$option_name', '$option_value', 'no')"; |
||
| 72 | } |
||
| 73 | |||
| 74 | $rows_added = $wpdb->query( $query . join( ',', $rows ) ); |
||
| 75 | |||
| 76 | if ( $rows_added !== count( $items ) ) { |
||
| 77 | return new WP_Error( 'row_count_mismatch', "The number of rows inserted didn't match the size of the input array" ); |
||
| 78 | } |
||
| 79 | } |
||
| 80 | |||
| 81 | // Peek at the front-most item on the queue without checking it out |
||
| 82 | function peek( $count = 1 ) { |
||
| 83 | $items = $this->fetch_items( $count ); |
||
| 84 | if ( $items ) { |
||
| 85 | return Jetpack_Sync_Utils::get_item_values( $items ); |
||
| 86 | } |
||
| 87 | |||
| 88 | return array(); |
||
| 89 | } |
||
| 90 | |||
| 91 | // lag is the difference in time between the age of the oldest item and the current time |
||
| 92 | function lag() { |
||
| 93 | global $wpdb; |
||
| 94 | |||
| 95 | $last_item_name = $wpdb->get_var( $wpdb->prepare( |
||
| 96 | "SELECT option_name FROM $wpdb->options WHERE option_name LIKE %s ORDER BY option_name ASC LIMIT 1", |
||
| 97 | "jpsq_{$this->id}-%" |
||
| 98 | ) ); |
||
| 99 | |||
| 100 | if ( ! $last_item_name ) { |
||
| 101 | return null; |
||
| 102 | } |
||
| 103 | |||
| 104 | // break apart the item name to get the timestamp |
||
| 105 | $matches = null; |
||
| 106 | if ( preg_match( '/^jpsq_' . $this->id . '-(\d+\.\d+)-/', $last_item_name, $matches ) ) { |
||
| 107 | return microtime( true ) - floatval( $matches[1] ); |
||
| 108 | } else { |
||
| 109 | return null; |
||
| 110 | } |
||
| 111 | } |
||
| 112 | |||
| 113 | function reset() { |
||
| 114 | global $wpdb; |
||
| 115 | $this->delete_checkout_id(); |
||
| 116 | $wpdb->query( $wpdb->prepare( |
||
| 117 | "DELETE FROM $wpdb->options WHERE option_name LIKE %s", "jpsq_{$this->id}-%" |
||
| 118 | ) ); |
||
| 119 | } |
||
| 120 | |||
| 121 | function size() { |
||
| 122 | global $wpdb; |
||
| 123 | |||
| 124 | return (int) $wpdb->get_var( $wpdb->prepare( |
||
| 125 | "SELECT count(*) FROM $wpdb->options WHERE option_name LIKE %s", "jpsq_{$this->id}-%" |
||
| 126 | ) ); |
||
| 127 | } |
||
| 128 | |||
| 129 | // we use this peculiar implementation because it's much faster than count(*) |
||
| 130 | function has_any_items() { |
||
| 131 | global $wpdb; |
||
| 132 | $value = $wpdb->get_var( $wpdb->prepare( |
||
| 133 | "SELECT exists( SELECT option_name FROM $wpdb->options WHERE option_name LIKE %s )", "jpsq_{$this->id}-%" |
||
| 134 | ) ); |
||
| 135 | |||
| 136 | return ( $value === "1" ); |
||
| 137 | } |
||
| 138 | |||
| 139 | function checkout( $buffer_size ) { |
||
| 140 | if ( $this->get_checkout_id() ) { |
||
| 141 | return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' ); |
||
| 142 | } |
||
| 143 | |||
| 144 | $buffer_id = uniqid(); |
||
| 145 | |||
| 146 | $result = $this->set_checkout_id( $buffer_id ); |
||
| 147 | |||
| 148 | if ( ! $result || is_wp_error( $result ) ) { |
||
| 149 | return $result; |
||
| 150 | } |
||
| 151 | |||
| 152 | $items = $this->fetch_items( $buffer_size ); |
||
| 153 | |||
| 154 | if ( count( $items ) === 0 ) { |
||
| 155 | return false; |
||
| 156 | } |
||
| 157 | |||
| 158 | $buffer = new Jetpack_Sync_Queue_Buffer( $buffer_id, array_slice( $items, 0, $buffer_size ) ); |
||
| 159 | |||
| 160 | return $buffer; |
||
| 161 | } |
||
| 162 | |||
| 163 | // this checks out rows until it either empties the queue or hits a certain memory limit |
||
| 164 | // it loads the sizes from the DB first so that it doesn't accidentally |
||
| 165 | // load more data into memory than it needs to. |
||
| 166 | // The only way it will load more items than $max_size is if a single queue item |
||
| 167 | // exceeds the memory limit, but in that case it will send that item by itself. |
||
| 168 | function checkout_with_memory_limit( $max_memory, $max_buffer_size = 500 ) { |
||
| 169 | if ( $this->get_checkout_id() ) { |
||
| 170 | return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' ); |
||
| 171 | } |
||
| 172 | |||
| 173 | $buffer_id = uniqid(); |
||
| 174 | |||
| 175 | $result = $this->set_checkout_id( $buffer_id ); |
||
| 176 | |||
| 177 | if ( ! $result || is_wp_error( $result ) ) { |
||
| 178 | return $result; |
||
| 179 | } |
||
| 180 | |||
| 181 | // get the map of buffer_id -> memory_size |
||
| 182 | global $wpdb; |
||
| 183 | |||
| 184 | $items_with_size = $wpdb->get_results( |
||
| 185 | $wpdb->prepare( |
||
| 186 | "SELECT option_name AS id, LENGTH(option_value) AS value_size FROM $wpdb->options WHERE option_name LIKE %s ORDER BY option_name ASC LIMIT %d", |
||
| 187 | "jpsq_{$this->id}-%", |
||
| 188 | $max_buffer_size |
||
| 189 | ), |
||
| 190 | OBJECT |
||
| 191 | ); |
||
| 192 | |||
| 193 | $total_memory = 0; |
||
| 194 | $item_ids = array(); |
||
| 195 | |||
| 196 | foreach ( $items_with_size as $item_with_size ) { |
||
| 197 | $total_memory += $item_with_size->value_size; |
||
| 198 | |||
| 199 | // if this is the first item and it exceeds memory, allow loop to continue |
||
| 200 | // we will exit on the next iteration instead |
||
| 201 | if ( $total_memory > $max_memory && count( $item_ids ) > 0 ) { |
||
| 202 | break; |
||
| 203 | } |
||
| 204 | $item_ids[] = $item_with_size->id; |
||
| 205 | } |
||
| 206 | |||
| 207 | $items = $this->fetch_items_by_id( $item_ids ); |
||
| 208 | |||
| 209 | if ( count( $items ) === 0 ) { |
||
| 210 | $this->delete_checkout_id(); |
||
| 211 | |||
| 212 | return false; |
||
| 213 | } |
||
| 214 | |||
| 215 | $buffer = new Jetpack_Sync_Queue_Buffer( $buffer_id, $items ); |
||
| 216 | |||
| 217 | return $buffer; |
||
| 218 | } |
||
| 219 | |||
| 220 | function checkin( $buffer ) { |
||
| 221 | $is_valid = $this->validate_checkout( $buffer ); |
||
| 222 | |||
| 223 | if ( is_wp_error( $is_valid ) ) { |
||
| 224 | return $is_valid; |
||
| 225 | } |
||
| 226 | |||
| 227 | $this->delete_checkout_id(); |
||
| 228 | |||
| 229 | return true; |
||
| 230 | } |
||
| 231 | |||
| 232 | function close( $buffer, $ids_to_remove = null ) { |
||
| 233 | $is_valid = $this->validate_checkout( $buffer ); |
||
| 234 | |||
| 235 | if ( is_wp_error( $is_valid ) ) { |
||
| 236 | return $is_valid; |
||
| 237 | } |
||
| 238 | |||
| 239 | $this->delete_checkout_id(); |
||
| 240 | |||
| 241 | // by default clear all items in the buffer |
||
| 242 | if ( is_null( $ids_to_remove ) ) { |
||
| 243 | $ids_to_remove = $buffer->get_item_ids(); |
||
| 244 | } |
||
| 245 | |||
| 246 | global $wpdb; |
||
| 247 | |||
| 248 | if ( count( $ids_to_remove ) > 0 ) { |
||
| 249 | $sql = "DELETE FROM $wpdb->options WHERE option_name IN (" . implode( ', ', array_fill( 0, count( $ids_to_remove ), '%s' ) ) . ')'; |
||
| 250 | $query = call_user_func_array( array( $wpdb, 'prepare' ), array_merge( array( $sql ), $ids_to_remove ) ); |
||
| 251 | $wpdb->query( $query ); |
||
| 252 | } |
||
| 253 | |||
| 254 | return true; |
||
| 255 | } |
||
| 256 | |||
| 257 | function flush_all() { |
||
| 258 | $items = Jetpack_Sync_Utils::get_item_values( $this->fetch_items() ); |
||
| 259 | $this->reset(); |
||
| 260 | |||
| 261 | return $items; |
||
| 262 | } |
||
| 263 | |||
| 264 | function get_all() { |
||
| 265 | return $this->fetch_items(); |
||
| 266 | } |
||
| 267 | |||
| 268 | // use with caution, this could allow multiple processes to delete |
||
| 269 | // and send from the queue at the same time |
||
| 270 | function force_checkin() { |
||
| 271 | $this->delete_checkout_id(); |
||
| 272 | } |
||
| 273 | |||
| 274 | // used to lock checkouts from the queue. |
||
| 275 | // tries to wait up to $timeout seconds for the queue to be empty |
||
| 276 | function lock( $timeout = 30 ) { |
||
| 277 | $tries = 0; |
||
| 278 | |||
| 279 | while ( $this->has_any_items() && $tries < $timeout ) { |
||
| 280 | sleep( 1 ); |
||
| 281 | $tries += 1; |
||
| 282 | } |
||
| 283 | |||
| 284 | if ( $tries === 30 ) { |
||
| 285 | return new WP_Error( 'lock_timeout', 'Timeout waiting for sync queue to empty' ); |
||
| 286 | } |
||
| 287 | |||
| 288 | if ( $this->get_checkout_id() ) { |
||
| 289 | return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' ); |
||
| 290 | } |
||
| 291 | |||
| 292 | // hopefully this means we can acquire a checkout? |
||
| 293 | $result = $this->set_checkout_id( 'lock' ); |
||
| 294 | |||
| 295 | if ( ! $result || is_wp_error( $result ) ) { |
||
| 296 | return $result; |
||
| 297 | } |
||
| 298 | |||
| 299 | return true; |
||
| 300 | } |
||
| 301 | |||
| 302 | function unlock() { |
||
| 303 | $this->delete_checkout_id(); |
||
| 304 | } |
||
| 305 | |||
| 306 | private function get_checkout_id() { |
||
| 307 | return get_transient( $this->get_checkout_transient_name() ); |
||
| 308 | } |
||
| 309 | |||
| 310 | private function set_checkout_id( $checkout_id ) { |
||
| 311 | return set_transient( $this->get_checkout_transient_name(), $checkout_id, 5 * 60 ); // 5 minute timeout |
||
| 312 | } |
||
| 313 | |||
| 314 | private function delete_checkout_id() { |
||
| 315 | delete_transient( $this->get_checkout_transient_name() ); |
||
| 316 | } |
||
| 317 | |||
| 318 | private function get_checkout_transient_name() { |
||
| 319 | return "jpsq_{$this->id}_checkout"; |
||
| 320 | } |
||
| 321 | |||
| 322 | private function get_next_data_row_option_name() { |
||
| 323 | // this option is specifically chosen to, as much as possible, preserve time order |
||
| 324 | // and minimise the possibility of collisions between multiple processes working |
||
| 325 | // at the same time |
||
| 326 | // TODO: confirm we only need to support PHP 5.05+ (otherwise we'll need to emulate microtime as float, and avoid PHP_INT_MAX) |
||
|
0 ignored issues
–
show
|
|||
| 327 | // @see: http://php.net/manual/en/function.microtime.php |
||
| 328 | $timestamp = sprintf( '%.6f', microtime( true ) ); |
||
| 329 | |||
| 330 | // row iterator is used to avoid collisions where we're writing data waaay fast in a single process |
||
| 331 | if ( $this->row_iterator === PHP_INT_MAX ) { |
||
| 332 | $this->row_iterator = 0; |
||
| 333 | } else { |
||
| 334 | $this->row_iterator += 1; |
||
| 335 | } |
||
| 336 | |||
| 337 | return 'jpsq_' . $this->id . '-' . $timestamp . '-' . getmypid() . '-' . $this->row_iterator; |
||
| 338 | } |
||
| 339 | |||
| 340 | private function fetch_items( $limit = null ) { |
||
| 341 | global $wpdb; |
||
| 342 | |||
| 343 | if ( $limit ) { |
||
| 344 | $query_sql = $wpdb->prepare( "SELECT option_name AS id, option_value AS value FROM $wpdb->options WHERE option_name LIKE %s ORDER BY option_name ASC LIMIT %d", "jpsq_{$this->id}-%", $limit ); |
||
| 345 | } else { |
||
| 346 | $query_sql = $wpdb->prepare( "SELECT option_name AS id, option_value AS value FROM $wpdb->options WHERE option_name LIKE %s ORDER BY option_name ASC", "jpsq_{$this->id}-%" ); |
||
| 347 | } |
||
| 348 | |||
| 349 | $items = $wpdb->get_results( $query_sql, OBJECT ); |
||
| 350 | foreach ( $items as $item ) { |
||
| 351 | $item->value = maybe_unserialize( $item->value ); |
||
| 352 | } |
||
| 353 | |||
| 354 | return $items; |
||
| 355 | } |
||
| 356 | |||
| 357 | private function fetch_items_by_id( $item_ids ) { |
||
| 358 | global $wpdb; |
||
| 359 | |||
| 360 | if ( count( $item_ids ) > 0 ) { |
||
| 361 | $sql = "SELECT option_name AS id, option_value AS value FROM $wpdb->options WHERE option_name IN (" . implode( ', ', array_fill( 0, count( $item_ids ), '%s' ) ) . ') ORDER BY option_name ASC'; |
||
| 362 | $query = call_user_func_array( array( $wpdb, 'prepare' ), array_merge( array( $sql ), $item_ids ) ); |
||
| 363 | $items = $wpdb->get_results( $query, OBJECT ); |
||
| 364 | foreach ( $items as $item ) { |
||
| 365 | $item->value = maybe_unserialize( $item->value ); |
||
| 366 | } |
||
| 367 | |||
| 368 | return $items; |
||
| 369 | } else { |
||
| 370 | return array(); |
||
| 371 | } |
||
| 372 | } |
||
| 373 | |||
| 374 | private function validate_checkout( $buffer ) { |
||
| 375 | if ( ! $buffer instanceof Jetpack_Sync_Queue_Buffer ) { |
||
| 376 | return new WP_Error( 'not_a_buffer', 'You must checkin an instance of Jetpack_Sync_Queue_Buffer' ); |
||
| 377 | } |
||
| 378 | |||
| 379 | $checkout_id = $this->get_checkout_id(); |
||
| 380 | |||
| 381 | if ( ! $checkout_id ) { |
||
| 382 | return new WP_Error( 'buffer_not_checked_out', 'There are no checked out buffers' ); |
||
| 383 | } |
||
| 384 | |||
| 385 | if ( $checkout_id != $buffer->id ) { |
||
| 386 | return new WP_Error( 'buffer_mismatch', 'The buffer you checked in was not checked out' ); |
||
| 387 | } |
||
| 388 | |||
| 389 | return true; |
||
| 390 | } |
||
| 391 | } |
||
| 392 | |||
| 393 | class Jetpack_Sync_Utils { |
||
|
0 ignored issues
–
show
|
|||
| 394 | |||
| 395 | static function get_item_values( $items ) { |
||
| 396 | return array_map( array( __CLASS__, 'get_item_value' ), $items ); |
||
| 397 | } |
||
| 398 | |||
| 399 | static function get_item_ids( $items ) { |
||
| 400 | return array_map( array( __CLASS__, 'get_item_id' ), $items ); |
||
| 401 | } |
||
| 402 | |||
| 403 | static private function get_item_value( $item ) { |
||
|
0 ignored issues
–
show
|
|||
| 404 | return $item->value; |
||
| 405 | } |
||
| 406 | |||
| 407 | static private function get_item_id( $item ) { |
||
|
0 ignored issues
–
show
|
|||
| 408 | return $item->id; |
||
| 409 | } |
||
| 410 | } |
||
| 411 |
Having each class in a dedicated file usually plays nice with PSR autoloaders and is therefore a well established practice. If you use other autoloaders, you might not want to follow this rule.