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 (message.isRecievedByDFBridge()) { 114 // expected for messages pending acks with kahadb.concurrentStoreAndDispatchQueues=true 115 if (LOG.isTraceEnabled()) { 116 LOG.trace("{} store replayed pending message due to concurrentStoreAndDispatchQueues {} seq: {}", this, message.getMessageId(), message.getMessageId().getFutureOrSequenceLong()); 117 } 118 } else { 119 LOG.warn("{} - cursor got duplicate from store {} seq: {}", this, message.getMessageId(), message.getMessageId().getFutureOrSequenceLong()); 120 duplicate(message); 121 } 122 } else { 123 LOG.warn("{} - cursor got duplicate send {} seq: {}", this, message.getMessageId(), message.getMessageId().getFutureOrSequenceLong()); 124 if (message.getMessageId().getEntryLocator() instanceof Long) { 125 // JDBC will store a duplicate (with new sequence id) - it needs an ack (AMQ4952Test) 126 duplicate(message); 127 } 128 } 129 return recovered; 130 } 131 132 // track for processing outside of store index lock so we can dlq 133 final LinkedList<Message> duplicatesFromStore = new LinkedList<Message>(); 134 private void duplicate(Message message) { 135 duplicatesFromStore.add(message); 136 } 137 138 void dealWithDuplicates() { 139 for (Message message : duplicatesFromStore) { 140 regionDestination.duplicateFromStore(message, getSubscription()); 141 } 142 duplicatesFromStore.clear(); 143 } 144 145 @Override 146 public final synchronized void reset() { 147 if (batchList.isEmpty()) { 148 try { 149 fillBatch(); 150 } catch (Exception e) { 151 LOG.error("{} - Failed to fill batch", this, e); 152 throw new RuntimeException(e); 153 } 154 } 155 clearIterator(true); 156 size(); 157 } 158 159 160 @Override 161 public synchronized void release() { 162 clearIterator(false); 163 } 164 165 private synchronized void clearIterator(boolean ensureIterator) { 166 boolean haveIterator = this.iterator != null; 167 this.iterator=null; 168 if(haveIterator&&ensureIterator) { 169 ensureIterator(); 170 } 171 } 172 173 private synchronized void ensureIterator() { 174 if(this.iterator==null) { 175 this.iterator=this.batchList.iterator(); 176 } 177 } 178 179 180 public final void finished() { 181 } 182 183 184 @Override 185 public final synchronized boolean hasNext() { 186 if (batchList.isEmpty()) { 187 try { 188 fillBatch(); 189 } catch (Exception e) { 190 LOG.error("{} - Failed to fill batch", this, e); 191 throw new RuntimeException(e); 192 } 193 } 194 ensureIterator(); 195 return this.iterator.hasNext(); 196 } 197 198 199 @Override 200 public final synchronized MessageReference next() { 201 MessageReference result = null; 202 if (!this.batchList.isEmpty()&&this.iterator.hasNext()) { 203 result = this.iterator.next(); 204 } 205 last = result; 206 if (result != null) { 207 result.incrementReferenceCount(); 208 } 209 return result; 210 } 211 212 @Override 213 public synchronized boolean tryAddMessageLast(MessageReference node, long wait) throws Exception { 214 boolean disableCache = false; 215 if (hasSpace()) { 216 if (!isCacheEnabled() && size==0 && isStarted() && useCache) { 217 if (LOG.isTraceEnabled()) { 218 LOG.trace("{} - enabling cache for empty store {} {}", this, node.getMessageId(), node.getMessageId().getFutureOrSequenceLong()); 219 } 220 setCacheEnabled(true); 221 } 222 if (isCacheEnabled()) { 223 if (recoverMessage(node.getMessage(),true)) { 224 trackLastCached(node); 225 } else { 226 dealWithDuplicates(); 227 return false; 228 } 229 } 230 } else { 231 disableCache = true; 232 } 233 234 if (disableCache && isCacheEnabled()) { 235 if (LOG.isTraceEnabled()) { 236 LOG.trace("{} - disabling cache on add {} {}", this, node.getMessageId(), node.getMessageId().getFutureOrSequenceLong()); 237 } 238 syncWithStore(node.getMessage()); 239 setCacheEnabled(false); 240 } 241 size++; 242 return true; 243 } 244 245 private void syncWithStore(Message currentAdd) throws Exception { 246 pruneLastCached(); 247 if (lastCachedIds[SYNC_ADD] == null) { 248 // possibly only async adds, lets wait on the potential last add and reset from there 249 for (ListIterator<MessageId> it = pendingCachedIds.listIterator(pendingCachedIds.size()); it.hasPrevious(); ) { 250 MessageId lastPending = it.previous(); 251 Object futureOrLong = lastPending.getFutureOrSequenceLong(); 252 if (futureOrLong instanceof Future) { 253 Future future = (Future) futureOrLong; 254 if (future.isCancelled()) { 255 continue; 256 } 257 try { 258 future.get(5, TimeUnit.SECONDS); 259 setLastCachedId(ASYNC_ADD, lastPending); 260 } catch (CancellationException ok) { 261 continue; 262 } catch (TimeoutException potentialDeadlock) { 263 LOG.debug("{} timed out waiting for async add", this, potentialDeadlock); 264 } catch (Exception worstCaseWeReplay) { 265 LOG.debug("{} exception waiting for async add", this, worstCaseWeReplay); 266 } 267 } else { 268 setLastCachedId(ASYNC_ADD, lastPending); 269 } 270 break; 271 } 272 if (lastCachedIds[ASYNC_ADD] != null) { 273 // ensure we don't skip current possibly sync add b/c we waited on the future 274 if (isAsync(currentAdd) || Long.compare(((Long) currentAdd.getMessageId().getFutureOrSequenceLong()), ((Long) lastCachedIds[ASYNC_ADD].getFutureOrSequenceLong())) > 0) { 275 setBatch(lastCachedIds[ASYNC_ADD]); 276 } 277 } 278 } else { 279 setBatch(lastCachedIds[SYNC_ADD]); 280 } 281 // cleanup 282 lastCachedIds[SYNC_ADD] = lastCachedIds[ASYNC_ADD] = null; 283 pendingCachedIds.clear(); 284 } 285 286 private void trackLastCached(MessageReference node) { 287 if (isAsync(node.getMessage())) { 288 pruneLastCached(); 289 pendingCachedIds.add(node.getMessageId()); 290 } else { 291 setLastCachedId(SYNC_ADD, node.getMessageId()); 292 } 293 } 294 295 private static final boolean isAsync(Message message) { 296 return message.isRecievedByDFBridge() || message.getMessageId().getFutureOrSequenceLong() instanceof Future; 297 } 298 299 private void pruneLastCached() { 300 for (Iterator<MessageId> it = pendingCachedIds.iterator(); it.hasNext(); ) { 301 MessageId candidate = it.next(); 302 final Object futureOrLong = candidate.getFutureOrSequenceLong(); 303 if (futureOrLong instanceof Future) { 304 Future future = (Future) futureOrLong; 305 if (future.isCancelled()) { 306 it.remove(); 307 } else { 308 // we don't want to wait for work to complete 309 break; 310 } 311 } else { 312 // complete 313 setLastCachedId(ASYNC_ADD, candidate); 314 315 // keep lock step with sync adds while order is preserved 316 if (lastCachedIds[SYNC_ADD] != null) { 317 long next = 1 + (Long)lastCachedIds[SYNC_ADD].getFutureOrSequenceLong(); 318 if (Long.compare((Long)futureOrLong, next) == 0) { 319 setLastCachedId(SYNC_ADD, candidate); 320 } 321 } 322 it.remove(); 323 } 324 } 325 } 326 327 private void setLastCachedId(final int index, MessageId candidate) { 328 MessageId lastCacheId = lastCachedIds[index]; 329 if (lastCacheId == null) { 330 lastCachedIds[index] = candidate; 331 } else { 332 Object lastCacheFutureOrSequenceLong = lastCacheId.getFutureOrSequenceLong(); 333 Object candidateOrSequenceLong = candidate.getFutureOrSequenceLong(); 334 if (lastCacheFutureOrSequenceLong == null) { // possibly null for topics 335 lastCachedIds[index] = candidate; 336 } else if (candidateOrSequenceLong != null && 337 Long.compare(((Long) candidateOrSequenceLong), ((Long) lastCacheFutureOrSequenceLong)) > 0) { 338 lastCachedIds[index] = candidate; 339 } 340 } 341 } 342 343 protected void setBatch(MessageId messageId) throws Exception { 344 } 345 346 347 @Override 348 public synchronized void addMessageFirst(MessageReference node) throws Exception { 349 setCacheEnabled(false); 350 size++; 351 } 352 353 354 @Override 355 public final synchronized void remove() { 356 size--; 357 if (iterator!=null) { 358 iterator.remove(); 359 } 360 if (last != null) { 361 last.decrementReferenceCount(); 362 } 363 } 364 365 366 @Override 367 public final synchronized void remove(MessageReference node) { 368 if (batchList.remove(node) != null) { 369 size--; 370 setCacheEnabled(false); 371 } 372 } 373 374 375 @Override 376 public final synchronized void clear() { 377 gc(); 378 } 379 380 381 @Override 382 public synchronized void gc() { 383 for (MessageReference msg : batchList) { 384 rollback(msg.getMessageId()); 385 msg.decrementReferenceCount(); 386 } 387 batchList.clear(); 388 clearIterator(false); 389 batchResetNeeded = true; 390 setCacheEnabled(false); 391 } 392 393 @Override 394 protected final synchronized void fillBatch() { 395 if (LOG.isTraceEnabled()) { 396 LOG.trace("{} fillBatch", this); 397 } 398 if (batchResetNeeded) { 399 resetSize(); 400 setMaxBatchSize(Math.min(regionDestination.getMaxPageSize(), size)); 401 resetBatch(); 402 this.batchResetNeeded = false; 403 } 404 if (this.batchList.isEmpty() && this.size >0) { 405 try { 406 doFillBatch(); 407 } catch (Exception e) { 408 LOG.error("{} - Failed to fill batch", this, e); 409 throw new RuntimeException(e); 410 } 411 } 412 } 413 414 415 @Override 416 public final synchronized boolean isEmpty() { 417 // negative means more messages added to store through queue.send since last reset 418 return size == 0; 419 } 420 421 422 @Override 423 public final synchronized boolean hasMessagesBufferedToDeliver() { 424 return !batchList.isEmpty(); 425 } 426 427 428 @Override 429 public final synchronized int size() { 430 if (size < 0) { 431 this.size = getStoreSize(); 432 } 433 return size; 434 } 435 436 @Override 437 public final synchronized long messageSize() { 438 return getStoreMessageSize(); 439 } 440 441 @Override 442 public String toString() { 443 return super.toString() + ":" + regionDestination.getActiveMQDestination().getPhysicalName() + ",batchResetNeeded=" + batchResetNeeded 444 + ",size=" + this.size + ",cacheEnabled=" + isCacheEnabled() 445 + ",maxBatchSize:" + maxBatchSize + ",hasSpace:" + hasSpace() + ",pendingCachedIds.size:" + pendingCachedIds.size() 446 + ",lastSyncCachedId:" + lastCachedIds[SYNC_ADD] + ",lastSyncCachedId-seq:" + (lastCachedIds[SYNC_ADD] != null ? lastCachedIds[SYNC_ADD].getFutureOrSequenceLong() : "null") 447 + ",lastAsyncCachedId:" + lastCachedIds[ASYNC_ADD] + ",lastAsyncCachedId-seq:" + (lastCachedIds[ASYNC_ADD] != null ? lastCachedIds[ASYNC_ADD].getFutureOrSequenceLong() : "null"); 448 } 449 450 protected abstract void doFillBatch() throws Exception; 451 452 protected abstract void resetBatch(); 453 454 protected abstract int getStoreSize(); 455 456 protected abstract long getStoreMessageSize(); 457 458 protected abstract boolean isStoreEmpty(); 459 460 public Subscription getSubscription() { 461 return null; 462 } 463}