001/** 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.activemq.broker.region.cursors; 018 019import java.util.Iterator; 020import java.util.LinkedList; 021import java.util.ListIterator; 022import java.util.concurrent.CancellationException; 023import java.util.concurrent.Future; 024import java.util.concurrent.TimeUnit; 025import java.util.concurrent.TimeoutException; 026 027import org.apache.activemq.broker.region.Destination; 028import org.apache.activemq.broker.region.MessageReference; 029import org.apache.activemq.broker.region.Subscription; 030import org.apache.activemq.command.Message; 031import org.apache.activemq.command.MessageId; 032import org.apache.activemq.store.MessageRecoveryListener; 033import org.slf4j.Logger; 034import org.slf4j.LoggerFactory; 035 036/** 037 * Store based cursor 038 * 039 */ 040public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor implements MessageRecoveryListener { 041 private static final Logger LOG = LoggerFactory.getLogger(AbstractStoreCursor.class); 042 protected final Destination regionDestination; 043 protected final PendingList batchList; 044 private Iterator<MessageReference> iterator = null; 045 protected boolean batchResetNeeded = false; 046 protected int size; 047 private final LinkedList<MessageId> pendingCachedIds = new LinkedList<>(); 048 private static int SYNC_ADD = 0; 049 private static int ASYNC_ADD = 1; 050 final MessageId[] lastCachedIds = new MessageId[2]; 051 protected boolean hadSpace = false; 052 053 054 055 protected AbstractStoreCursor(Destination destination) { 056 super((destination != null ? destination.isPrioritizedMessages():false)); 057 this.regionDestination=destination; 058 if (this.prioritizedMessages) { 059 this.batchList= new PrioritizedPendingList(); 060 } else { 061 this.batchList = new OrderedPendingList(); 062 } 063 } 064 065 066 @Override 067 public final synchronized void start() throws Exception{ 068 if (!isStarted()) { 069 super.start(); 070 resetBatch(); 071 resetSize(); 072 setCacheEnabled(size==0&&useCache); 073 } 074 } 075 076 protected void resetSize() { 077 this.size = getStoreSize(); 078 } 079 080 @Override 081 public void rebase() { 082 resetSize(); 083 } 084 085 @Override 086 public final synchronized void stop() throws Exception { 087 resetBatch(); 088 super.stop(); 089 gc(); 090 } 091 092 093 @Override 094 public final boolean recoverMessage(Message message) throws Exception { 095 return recoverMessage(message,false); 096 } 097 098 public synchronized boolean recoverMessage(Message message, boolean cached) throws Exception { 099 boolean recovered = false; 100 message.setRegionDestination(regionDestination); 101 if (recordUniqueId(message.getMessageId())) { 102 if (!cached) { 103 if( message.getMemoryUsage()==null ) { 104 message.setMemoryUsage(this.getSystemUsage().getMemoryUsage()); 105 } 106 } 107 message.incrementReferenceCount(); 108 batchList.addMessageLast(message); 109 clearIterator(true); 110 recovered = true; 111 } else if (!cached) { 112 // a duplicate from the store (!cached) - needs to be removed/acked - otherwise it will get re dispatched on restart 113 if (duplicateFromStoreExcepted(message)) { 114 if (LOG.isTraceEnabled()) { 115 LOG.trace("{} store replayed pending message due to concurrentStoreAndDispatchQueues {} seq: {}", this, message.getMessageId(), message.getMessageId().getFutureOrSequenceLong()); 116 } 117 } else { 118 LOG.warn("{} - cursor got duplicate from store {} seq: {}", this, message.getMessageId(), message.getMessageId().getFutureOrSequenceLong()); 119 duplicate(message); 120 } 121 } else { 122 LOG.warn("{} - cursor got duplicate send {} seq: {}", this, message.getMessageId(), message.getMessageId().getFutureOrSequenceLong()); 123 if (gotToTheStore(message)) { 124 duplicate(message); 125 } 126 } 127 return recovered; 128 } 129 130 protected boolean duplicateFromStoreExcepted(Message message) { 131 // expected for messages pending acks with kahadb.concurrentStoreAndDispatchQueues=true for 132 // which this existing unused flag has been repurposed 133 return message.isRecievedByDFBridge(); 134 } 135 136 public static boolean gotToTheStore(Message message) throws Exception { 137 if (message.isRecievedByDFBridge()) { 138 // concurrent store and dispatch - wait to see if the message gets to the store to see 139 // if the index suppressed it (original still present), or whether it was stored and needs to be removed 140 Object possibleFuture = message.getMessageId().getFutureOrSequenceLong(); 141 if (possibleFuture instanceof Future) { 142 ((Future) possibleFuture).get(); 143 } 144 // need to access again after wait on future 145 Object sequence = message.getMessageId().getFutureOrSequenceLong(); 146 return (sequence != null && sequence instanceof Long && Long.compare((Long) sequence, -1l) != 0); 147 } 148 return true; 149 } 150 151 // track for processing outside of store index lock so we can dlq 152 final LinkedList<Message> duplicatesFromStore = new LinkedList<Message>(); 153 private void duplicate(Message message) { 154 duplicatesFromStore.add(message); 155 } 156 157 void dealWithDuplicates() { 158 for (Message message : duplicatesFromStore) { 159 regionDestination.duplicateFromStore(message, getSubscription()); 160 } 161 duplicatesFromStore.clear(); 162 } 163 164 @Override 165 public final synchronized void reset() { 166 if (batchList.isEmpty()) { 167 try { 168 fillBatch(); 169 } catch (Exception e) { 170 LOG.error("{} - Failed to fill batch", this, e); 171 throw new RuntimeException(e); 172 } 173 } 174 clearIterator(true); 175 size(); 176 } 177 178 179 @Override 180 public synchronized void release() { 181 clearIterator(false); 182 } 183 184 private synchronized void clearIterator(boolean ensureIterator) { 185 boolean haveIterator = this.iterator != null; 186 this.iterator=null; 187 if(haveIterator&&ensureIterator) { 188 ensureIterator(); 189 } 190 } 191 192 private synchronized void ensureIterator() { 193 if(this.iterator==null) { 194 this.iterator=this.batchList.iterator(); 195 } 196 } 197 198 199 public final void finished() { 200 } 201 202 203 @Override 204 public final synchronized boolean hasNext() { 205 if (batchList.isEmpty()) { 206 try { 207 fillBatch(); 208 } catch (Exception e) { 209 LOG.error("{} - Failed to fill batch", this, e); 210 throw new RuntimeException(e); 211 } 212 } 213 ensureIterator(); 214 return this.iterator.hasNext(); 215 } 216 217 218 @Override 219 public final synchronized MessageReference next() { 220 MessageReference result = null; 221 if (!this.batchList.isEmpty()&&this.iterator.hasNext()) { 222 result = this.iterator.next(); 223 } 224 last = result; 225 if (result != null) { 226 result.incrementReferenceCount(); 227 } 228 return result; 229 } 230 231 @Override 232 public synchronized boolean tryAddMessageLast(MessageReference node, long wait) throws Exception { 233 boolean disableCache = false; 234 if (hasSpace()) { 235 if (isCacheEnabled()) { 236 if (recoverMessage(node.getMessage(),true)) { 237 trackLastCached(node); 238 } else { 239 dealWithDuplicates(); 240 return false; 241 } 242 } 243 } else { 244 disableCache = true; 245 } 246 247 if (disableCache && isCacheEnabled()) { 248 if (LOG.isTraceEnabled()) { 249 LOG.trace("{} - disabling cache on add {} {}", this, node.getMessageId(), node.getMessageId().getFutureOrSequenceLong()); 250 } 251 syncWithStore(node.getMessage()); 252 setCacheEnabled(false); 253 } 254 size++; 255 return true; 256 } 257 258 @Override 259 public synchronized boolean isCacheEnabled() { 260 return super.isCacheEnabled() || enableCacheNow(); 261 } 262 263 protected boolean enableCacheNow() { 264 boolean result = false; 265 if (canEnableCash()) { 266 setCacheEnabled(true); 267 result = true; 268 if (LOG.isTraceEnabled()) { 269 LOG.trace("{} enabling cache on empty store", this); 270 } 271 } 272 return result; 273 } 274 275 protected boolean canEnableCash() { 276 return useCache && size==0 && hasSpace() && isStarted(); 277 } 278 279 private void syncWithStore(Message currentAdd) throws Exception { 280 pruneLastCached(); 281 for (ListIterator<MessageId> it = pendingCachedIds.listIterator(pendingCachedIds.size()); it.hasPrevious(); ) { 282 MessageId lastPending = it.previous(); 283 Object futureOrLong = lastPending.getFutureOrSequenceLong(); 284 if (futureOrLong instanceof Future) { 285 Future future = (Future) futureOrLong; 286 if (future.isCancelled()) { 287 continue; 288 } 289 try { 290 future.get(5, TimeUnit.SECONDS); 291 setLastCachedId(ASYNC_ADD, lastPending); 292 } catch (CancellationException ok) { 293 continue; 294 } catch (TimeoutException potentialDeadlock) { 295 LOG.debug("{} timed out waiting for async add", this, potentialDeadlock); 296 } catch (Exception worstCaseWeReplay) { 297 LOG.debug("{} exception waiting for async add", this, worstCaseWeReplay); 298 } 299 } else { 300 setLastCachedId(ASYNC_ADD, lastPending); 301 } 302 break; 303 } 304 305 MessageId candidate = lastCachedIds[ASYNC_ADD]; 306 if (candidate != null) { 307 // ensure we don't skip current possibly sync add b/c we waited on the future 308 if (!isAsync(currentAdd) && Long.compare(((Long) currentAdd.getMessageId().getFutureOrSequenceLong()), ((Long) lastCachedIds[ASYNC_ADD].getFutureOrSequenceLong())) < 0) { 309 if (LOG.isTraceEnabled()) { 310 LOG.trace("no set batch from async:" + candidate.getFutureOrSequenceLong() + " >= than current: " + currentAdd.getMessageId().getFutureOrSequenceLong() + ", " + this); 311 } 312 candidate = null; 313 } 314 } 315 if (candidate == null) { 316 candidate = lastCachedIds[SYNC_ADD]; 317 } 318 if (candidate != null) { 319 setBatch(candidate); 320 } 321 // cleanup 322 lastCachedIds[SYNC_ADD] = lastCachedIds[ASYNC_ADD] = null; 323 pendingCachedIds.clear(); 324 } 325 326 private void trackLastCached(MessageReference node) { 327 if (isAsync(node.getMessage())) { 328 pruneLastCached(); 329 pendingCachedIds.add(node.getMessageId()); 330 } else { 331 setLastCachedId(SYNC_ADD, node.getMessageId()); 332 } 333 } 334 335 private static final boolean isAsync(Message message) { 336 return message.isRecievedByDFBridge() || message.getMessageId().getFutureOrSequenceLong() instanceof Future; 337 } 338 339 private void pruneLastCached() { 340 for (Iterator<MessageId> it = pendingCachedIds.iterator(); it.hasNext(); ) { 341 MessageId candidate = it.next(); 342 final Object futureOrLong = candidate.getFutureOrSequenceLong(); 343 if (futureOrLong instanceof Future) { 344 Future future = (Future) futureOrLong; 345 if (future.isCancelled()) { 346 it.remove(); 347 } else { 348 // we don't want to wait for work to complete 349 break; 350 } 351 } else { 352 // complete 353 setLastCachedId(ASYNC_ADD, candidate); 354 355 // keep lock step with sync adds while order is preserved 356 if (lastCachedIds[SYNC_ADD] != null) { 357 long next = 1 + (Long)lastCachedIds[SYNC_ADD].getFutureOrSequenceLong(); 358 if (Long.compare((Long)futureOrLong, next) == 0) { 359 setLastCachedId(SYNC_ADD, candidate); 360 } 361 } 362 it.remove(); 363 } 364 } 365 } 366 367 private void setLastCachedId(final int index, MessageId candidate) { 368 MessageId lastCacheId = lastCachedIds[index]; 369 if (lastCacheId == null) { 370 lastCachedIds[index] = candidate; 371 } else { 372 Object lastCacheFutureOrSequenceLong = lastCacheId.getFutureOrSequenceLong(); 373 Object candidateOrSequenceLong = candidate.getFutureOrSequenceLong(); 374 if (lastCacheFutureOrSequenceLong == null) { // possibly null for topics 375 lastCachedIds[index] = candidate; 376 } else if (candidateOrSequenceLong != null && 377 Long.compare(((Long) candidateOrSequenceLong), ((Long) lastCacheFutureOrSequenceLong)) > 0) { 378 lastCachedIds[index] = candidate; 379 } if (LOG.isTraceEnabled()) { 380 LOG.trace("no set last cached[" + index + "] current:" + lastCacheFutureOrSequenceLong + " <= than candidate: " + candidateOrSequenceLong+ ", " + this); 381 } 382 } 383 } 384 385 protected void setBatch(MessageId messageId) throws Exception { 386 } 387 388 389 @Override 390 public synchronized void addMessageFirst(MessageReference node) throws Exception { 391 setCacheEnabled(false); 392 size++; 393 } 394 395 396 @Override 397 public final synchronized void remove() { 398 size--; 399 if (iterator!=null) { 400 iterator.remove(); 401 } 402 if (last != null) { 403 last.decrementReferenceCount(); 404 } 405 } 406 407 408 @Override 409 public final synchronized void remove(MessageReference node) { 410 if (batchList.remove(node) != null) { 411 size--; 412 setCacheEnabled(false); 413 } 414 } 415 416 417 @Override 418 public final synchronized void clear() { 419 gc(); 420 } 421 422 423 @Override 424 public synchronized void gc() { 425 for (MessageReference msg : batchList) { 426 rollback(msg.getMessageId()); 427 msg.decrementReferenceCount(); 428 } 429 batchList.clear(); 430 clearIterator(false); 431 batchResetNeeded = true; 432 setCacheEnabled(false); 433 } 434 435 @Override 436 protected final synchronized void fillBatch() { 437 if (LOG.isTraceEnabled()) { 438 LOG.trace("{} fillBatch", this); 439 } 440 if (batchResetNeeded) { 441 resetSize(); 442 setMaxBatchSize(Math.min(regionDestination.getMaxPageSize(), size)); 443 resetBatch(); 444 this.batchResetNeeded = false; 445 } 446 if (this.batchList.isEmpty() && this.size >0) { 447 try { 448 doFillBatch(); 449 } catch (Exception e) { 450 LOG.error("{} - Failed to fill batch", this, e); 451 throw new RuntimeException(e); 452 } 453 } 454 } 455 456 457 @Override 458 public final synchronized boolean isEmpty() { 459 // negative means more messages added to store through queue.send since last reset 460 return size == 0; 461 } 462 463 464 @Override 465 public final synchronized boolean hasMessagesBufferedToDeliver() { 466 return !batchList.isEmpty(); 467 } 468 469 470 @Override 471 public final synchronized int size() { 472 if (size < 0) { 473 this.size = getStoreSize(); 474 } 475 return size; 476 } 477 478 @Override 479 public final synchronized long messageSize() { 480 return getStoreMessageSize(); 481 } 482 483 @Override 484 public String toString() { 485 return super.toString() + ":" + regionDestination.getActiveMQDestination().getPhysicalName() + ",batchResetNeeded=" + batchResetNeeded 486 + ",size=" + this.size + ",cacheEnabled=" + isCacheEnabled() 487 + ",maxBatchSize:" + maxBatchSize + ",hasSpace:" + hasSpace() + ",pendingCachedIds.size:" + pendingCachedIds.size() 488 + ",lastSyncCachedId:" + lastCachedIds[SYNC_ADD] + ",lastSyncCachedId-seq:" + (lastCachedIds[SYNC_ADD] != null ? lastCachedIds[SYNC_ADD].getFutureOrSequenceLong() : "null") 489 + ",lastAsyncCachedId:" + lastCachedIds[ASYNC_ADD] + ",lastAsyncCachedId-seq:" + (lastCachedIds[ASYNC_ADD] != null ? lastCachedIds[ASYNC_ADD].getFutureOrSequenceLong() : "null"); 490 } 491 492 protected abstract void doFillBatch() throws Exception; 493 494 protected abstract void resetBatch(); 495 496 protected abstract int getStoreSize(); 497 498 protected abstract long getStoreMessageSize(); 499 500 protected abstract boolean isStoreEmpty(); 501 502 public Subscription getSubscription() { 503 return null; 504 } 505}