Package org.apache.activemq.artemis.core.paging.impl

Class PagingStoreImpl

java.lang.Object

org.apache.activemq.artemis.core.paging.impl.PagingStoreImpl

All Implemented Interfaces:

RefCountMessageListener, PagingStore, ActiveMQComponent

public class PagingStoreImpl
extends Object
implements PagingStore

See Also:

• PagingStore

Constructor Summary

Constructors

Constructor	Description
PagingStoreImpl(SimpleString address, ScheduledExecutorService scheduledExecutor, long syncTimeout, PagingManager pagingManager, StorageManager storageManager, SequentialFileFactory fileFactory, PagingStoreFactory storeFactory, SimpleString storeName, AddressSettings addressSettings, ArtemisExecutor executor, boolean syncNonTransactional)

Method Summary

Modifier and Type	Method	Description
void	addSize(int size, boolean sizeOnly, boolean affectGlobal)	Add size to this PageStore.
void	applySetting(AddressSettings addressSettings)
void	block()
boolean	checkMemory(boolean runOnFailure, Runnable runWhenAvailableParameter, Runnable runWhenBlocking, Consumer<AtomicRunnable> blockedCallback)
boolean	checkMemory(Runnable runWhenAvailable, Consumer<AtomicRunnable> blockedCallback)
boolean	checkPageFileExists(long pageNumber)
void	checkPageLimit(long numberOfMessages)
boolean	checkReleasedMemory()	It will return true if the destination is leaving blocking.
void	counterSnapshot()	Save snapshots on the counters
String	createFileName(long pageID)
protected PageTimedWriter	createPageTimedWriter(ScheduledExecutorService scheduledExecutor, long syncTimeout)
Page	depage()	Returns a Page out of the Page System without reading it.
void	destroy()
protected void	directWritePage(PagedMessage pagedMessage, boolean lineUp, boolean originalReplicated)
void	disableCleanup()	This method will disable cleanup of pages.
void	durableDown(Message message, int durableCount)
void	durableUp(Message message, int durableCount)
void	enableCleanup()	This method will re-enable cleanup of pages.
PagingStoreImpl	enforceAddressFullMessagePolicy(AddressFullMessagePolicy enforcedAddressFullMessagePolicy)
void	execute(Runnable run)
void	flushExecutors()	This is used mostly by tests.
void	forceAnotherPage()
void	forceAnotherPage(boolean useExecutor)
protected void	forEachUsedPage(Consumer<Page> consumerPage)
SimpleString	getAddress()
long	getAddressElements()
AddressFullMessagePolicy	getAddressFullMessagePolicy()
int	getAddressLimitPercent()
long	getAddressSize()
Collection<Integer>	getCurrentIds()	Files to synchronize with a remote backup.
Page	getCurrentPage()
long	getCurrentWritingPage()	Returns the page id of the current page in which the system is writing files.
PageCursorProvider	getCursorProvider()
ArtemisExecutor	getExecutor()
protected SequentialFileFactory	getFileFactory()
long	getFirstPage()
File	getFolder()
String	getFolderName()
int	getMaxPageReadBytes()
int	getMaxPageReadMessages()
long	getMaxSize()
int	getNumberOfFiles()
long	getNumberOfPages()
PageFullMessagePolicy	getPageFullMessagePolicy()
Long	getPageLimitBytes()
Long	getPageLimitMessages()
int	getPageSizeBytes()
PageTimedWriter	getPageTimedWriter()
PagingManager	getPagingManager()
int	getPrefetchPageBytes()
int	getPrefetchPageMessages()
StorageManager	getStorageManager()
SimpleString	getStoreName()
protected int	getUsedPagesSize()
boolean	hasPendingIO()
void	ioSync()	Performs a real sync on the current IO file.
boolean	isFull()
boolean	isPageFull()
boolean	isPaging()	This method will look if the current state of paging is not paging, without using a lock.
boolean	isRejectingMessages()
boolean	isStarted()
Page	newPageObject(long pageNumber)
boolean	page(Message message, Transaction tx, RouteContextList listCtx)	Write message to page if we are paging.
int	page(Message message, Transaction tx, RouteContextList listCtx, Function<Message,Message> pageDecorator, boolean useFlowControl)
void	pageFull(PageSubscription subscription)	Callback to be used by a counter when the Page is full for that counter
void	processReload()
void	readLock()
boolean	readLock(long timeout)
void	readUnlock()
void	refDown(Message message, int count)
void	refUp(Message message, int count)
protected void	reloadLivePage(long pageId)
Page	removePage(int pageId)	Returns a Page out of the Page System without reading it.
protected void	replacePagedTimedWriter(PageTimedWriter writer)
void	sendPages(ReplicationManager replicator, Collection<Integer> pageIds)	Sends the pages with given IDs to the ReplicationManager.
void	start()
boolean	startPaging()	Returns true if paging was started, or false if paging was already started before this call.
void	stop()
void	stopPaging()
String	toString()
void	unblock()
final Page	usePage(long pageId)
Page	usePage(long pageId, boolean create)	Use this method when you want to use the cache of used pages.
Page	usePage(long pageId, boolean createEntry, boolean createFile)
void	writeFlowControl(int credits)
void	writeLock()
boolean	writeLock(long timeout)	Write lock the PagingStore.
void	writeUnlock()	Releases locks acquired with PagingStore.writeLock(long).

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Methods inherited from interface org.apache.activemq.artemis.core.server.ActiveMQComponent

asyncStop

Methods inherited from interface org.apache.activemq.artemis.core.paging.PagingStore

addSize, addSize

Methods inherited from interface org.apache.activemq.artemis.api.core.RefCountMessageListener

usageDown, usageUp

Constructor Details

PagingStoreImpl

public PagingStoreImpl(SimpleString address,
 ScheduledExecutorService scheduledExecutor,
 long syncTimeout,
 PagingManager pagingManager,
 StorageManager storageManager,
 SequentialFileFactory fileFactory,
 PagingStoreFactory storeFactory,
 SimpleString storeName,
 AddressSettings addressSettings,
 ArtemisExecutor executor,
 boolean syncNonTransactional)

Method Details

createPageTimedWriter

protected PageTimedWriter createPageTimedWriter(ScheduledExecutorService scheduledExecutor,
 long syncTimeout)

replacePagedTimedWriter

protected void replacePagedTimedWriter(PageTimedWriter writer)

applySetting

public void applySetting(AddressSettings addressSettings)

Specified by:

applySetting in interface PagingStore

toString

public String toString()

Overrides:

toString in class Object

getPageFullMessagePolicy

public PageFullMessagePolicy getPageFullMessagePolicy()

Specified by:

getPageFullMessagePolicy in interface PagingStore

getPageLimitMessages

public Long getPageLimitMessages()

Specified by:

getPageLimitMessages in interface PagingStore

getPageLimitBytes

public Long getPageLimitBytes()

Specified by:

getPageLimitBytes in interface PagingStore

pageFull

public void pageFull(PageSubscription subscription)

Description copied from interface: PagingStore

Callback to be used by a counter when the Page is full for that counter

Specified by:

pageFull in interface PagingStore

isPageFull

public boolean isPageFull()

Specified by:

isPageFull in interface PagingStore

checkPageLimit

public void checkPageLimit(long numberOfMessages)

Specified by:

checkPageLimit in interface PagingStore

readLock

public void readLock()

Specified by:

readLock in interface PagingStore

readLock

public boolean readLock(long timeout)

Specified by:

readLock in interface PagingStore

readUnlock

public void readUnlock()

Specified by:

readUnlock in interface PagingStore

writeLock

public void writeLock()

Specified by:

writeLock in interface PagingStore

writeLock

public boolean writeLock(long timeout)

Description copied from interface: PagingStore

Write lock the PagingStore.

Specified by:

writeLock in interface PagingStore

Parameters:

timeout - milliseconds to wait for the lock. If value is -1 then wait indefinitely.

Returns:

true if the lock was obtained, false otherwise

writeUnlock

public void writeUnlock()

Description copied from interface: PagingStore

Releases locks acquired with PagingStore.writeLock(long).

Specified by:

writeUnlock in interface PagingStore

getCursorProvider

public PageCursorProvider getCursorProvider()

Specified by:

getCursorProvider in interface PagingStore

getFirstPage

public long getFirstPage()

Specified by:

getFirstPage in interface PagingStore

getAddress

public SimpleString getAddress()

Specified by:

getAddress in interface PagingStore

getAddressSize

public long getAddressSize()

Specified by:

getAddressSize in interface PagingStore

getAddressElements

public long getAddressElements()

Specified by:

getAddressElements in interface PagingStore

getMaxSize

public long getMaxSize()

Specified by:

getMaxSize in interface PagingStore

getMaxPageReadBytes

public int getMaxPageReadBytes()

Specified by:

getMaxPageReadBytes in interface PagingStore

getPrefetchPageBytes

public int getPrefetchPageBytes()

Specified by:

getPrefetchPageBytes in interface PagingStore

getMaxPageReadMessages

public int getMaxPageReadMessages()

Specified by:

getMaxPageReadMessages in interface PagingStore

getPrefetchPageMessages

public int getPrefetchPageMessages()

Specified by:

getPrefetchPageMessages in interface PagingStore

getAddressFullMessagePolicy

public AddressFullMessagePolicy getAddressFullMessagePolicy()

Specified by:

getAddressFullMessagePolicy in interface PagingStore

enforceAddressFullMessagePolicy

public PagingStoreImpl enforceAddressFullMessagePolicy(AddressFullMessagePolicy enforcedAddressFullMessagePolicy)

Specified by:

enforceAddressFullMessagePolicy in interface PagingStore

getPageSizeBytes

public int getPageSizeBytes()

Specified by:

getPageSizeBytes in interface PagingStore

getFolder

public File getFolder()

Specified by:

getFolder in interface PagingStore

getFolderName

public String getFolderName()

Specified by:

getFolderName in interface PagingStore

isPaging

public boolean isPaging()

Description copied from interface: PagingStore

This method will look if the current state of paging is not paging, without using a lock. For cases where you need absolutely atomic results, check it directly on the internal variables while requiring a readLock.

It's ok to look for this with an estimate on starting a task or not, but you will need to recheck on actual paging operations.

Specified by:

isPaging in interface PagingStore

getNumberOfPages

public long getNumberOfPages()

Specified by:

getNumberOfPages in interface PagingStore

getCurrentWritingPage

public long getCurrentWritingPage()

Description copied from interface: PagingStore

Returns the page id of the current page in which the system is writing files.

Specified by:

getCurrentWritingPage in interface PagingStore

Returns:

the page id of the current page in which the system is writing files

getStoreName

public SimpleString getStoreName()

Specified by:

getStoreName in interface PagingStore

ioSync

public void ioSync()
            throws Exception

Description copied from interface: PagingStore

Performs a real sync on the current IO file.

Specified by:

ioSync in interface PagingStore

Throws:

Exception

processReload

public void processReload()
                   throws Exception

Specified by:

processReload in interface PagingStore

Throws:

Exception

getPagingManager

public PagingManager getPagingManager()

Specified by:

getPagingManager in interface PagingStore

isStarted

public boolean isStarted()

Specified by:

isStarted in interface ActiveMQComponent

counterSnapshot

public void counterSnapshot()

Description copied from interface: PagingStore

Save snapshots on the counters

Specified by:

counterSnapshot in interface PagingStore

stop

public void stop()
          throws Exception

Specified by:

stop in interface ActiveMQComponent

Throws:

Exception

getExecutor

public ArtemisExecutor getExecutor()

Specified by:

getExecutor in interface PagingStore

execute

public void execute(Runnable run)

Specified by:

execute in interface PagingStore

flushExecutors

public void flushExecutors()

Description copied from interface: PagingStore

This is used mostly by tests. We will wait any pending runnable to finish its execution

Specified by:

flushExecutors in interface PagingStore

getNumberOfFiles

public int getNumberOfFiles()
                     throws Exception

Throws:

Exception

start

public void start()
           throws Exception

Specified by:

start in interface ActiveMQComponent

Throws:

Exception

reloadLivePage

protected void reloadLivePage(long pageId)
                       throws Exception

Throws:

Exception

stopPaging

public void stopPaging()

Specified by:

stopPaging in interface PagingStore

startPaging

public boolean startPaging()

Description copied from interface: PagingStore

Returns true if paging was started, or false if paging was already started before this call.

Specified by:

startPaging in interface PagingStore

Returns:

true if paging was started, or false if paging was already started before this call

getCurrentPage

public Page getCurrentPage()

Specified by:

getCurrentPage in interface PagingStore

checkPageFileExists

public boolean checkPageFileExists(long pageNumber)

Specified by:

checkPageFileExists in interface PagingStore

newPageObject

public Page newPageObject(long pageNumber)
                   throws Exception

Specified by:

newPageObject in interface PagingStore

Throws:

Exception

usePage

public final Page usePage(long pageId)

Specified by:

usePage in interface PagingStore

usePage

public Page usePage(long pageId,
 boolean create)

Description copied from interface: PagingStore

Use this method when you want to use the cache of used pages. If you are just using offline (e.g. print-data), use the newPageObject method.

Specified by:

usePage in interface PagingStore

usePage

public Page usePage(long pageId,
 boolean createEntry,
 boolean createFile)

Specified by:

usePage in interface PagingStore

getFileFactory

protected SequentialFileFactory getFileFactory()
                                        throws Exception

Throws:

Exception

forceAnotherPage

public void forceAnotherPage()
                      throws Exception

Specified by:

forceAnotherPage in interface PagingStore

Throws:

Exception

forceAnotherPage

public void forceAnotherPage(boolean useExecutor)
                      throws Exception

Specified by:

forceAnotherPage in interface PagingStore

Throws:

Exception

removePage

public Page removePage(int pageId)

Returns a Page out of the Page System without reading it.

The method calling this method will remove the page and will start reading it outside of any locks. This method could also replace the current file by a new file, and that process is done through acquiring a writeLock on currentPageLock.

Observation: This method is used internally as part of the regular depage process, but externally is used only on tests, and that's why this method is part of the Testable Interface

Specified by:

removePage in interface PagingStore

depage

public Page depage()
            throws Exception

Returns a Page out of the Page System without reading it.

The method calling this method will remove the page and will start reading it outside of any locks. This method could also replace the current file by a new file, and that process is done through acquiring a writeLock on currentPageLock.

Observation: This method is used internally as part of the regular depage process, but externally is used only on tests, and that's why this method is part of the Testable Interface

Specified by:

depage in interface PagingStore

Throws:

Exception - Note: This should still be part of the interface, even though ActiveMQ Artemis only uses through the

checkMemory

public boolean checkMemory(Runnable runWhenAvailable,
 Consumer<AtomicRunnable> blockedCallback)

Specified by:

checkMemory in interface PagingStore

checkMemory

public boolean checkMemory(boolean runOnFailure,
 Runnable runWhenAvailableParameter,
 Runnable runWhenBlocking,
 Consumer<AtomicRunnable> blockedCallback)

Specified by:

checkMemory in interface PagingStore

addSize

public void addSize(int size,
 boolean sizeOnly,
 boolean affectGlobal)

Description copied from interface: PagingStore

Add size to this PageStore.

Specified by:

addSize in interface PagingStore

sizeOnly - if false we won't increment the number of messages. (add references for example)

checkReleasedMemory

public boolean checkReleasedMemory()

Description copied from interface: PagingStore

It will return true if the destination is leaving blocking.

Specified by:

checkReleasedMemory in interface PagingStore

page

public boolean page(Message message,
 Transaction tx,
 RouteContextList listCtx)
             throws Exception

Description copied from interface: PagingStore

Write message to page if we are paging.

Specified by:

page in interface PagingStore

Returns:

true if we are paging and have handled the data, false if the data needs to be sent to the journal

Throws:

NullPointerException - if readLock is null

Exception

page

public int page(Message message,
 Transaction tx,
 RouteContextList listCtx,
 Function<Message,Message> pageDecorator,
 boolean useFlowControl)
         throws Exception

Specified by:

page in interface PagingStore

Throws:

Exception

writeFlowControl

public void writeFlowControl(int credits)

Specified by:

writeFlowControl in interface PagingStore

directWritePage

protected void directWritePage(PagedMessage pagedMessage,
 boolean lineUp,
 boolean originalReplicated)
                        throws Exception

Throws:

Exception

disableCleanup

public void disableCleanup()

This method will disable cleanup of pages. No page will be deleted after this call.

Specified by:

disableCleanup in interface PagingStore

enableCleanup

public void enableCleanup()

This method will re-enable cleanup of pages. Notice that it will also start cleanup threads.

Specified by:

enableCleanup in interface PagingStore

durableDown

public void durableDown(Message message,
 int durableCount)

Specified by:

durableDown in interface RefCountMessageListener

durableUp

public void durableUp(Message message,
 int durableCount)

Specified by:

durableUp in interface RefCountMessageListener

refUp

public void refUp(Message message,
 int count)

Specified by:

refUp in interface RefCountMessageListener

refDown

public void refDown(Message message,
 int count)

Specified by:

refDown in interface RefCountMessageListener

hasPendingIO

public boolean hasPendingIO()

Specified by:

hasPendingIO in interface PagingStore

getPageTimedWriter

public PageTimedWriter getPageTimedWriter()

destroy

public void destroy()
             throws Exception

Specified by:

destroy in interface PagingStore

Throws:

Exception

createFileName

public String createFileName(long pageID)

isFull

public boolean isFull()

Specified by:

isFull in interface PagingStore

getAddressLimitPercent

public int getAddressLimitPercent()

Specified by:

getAddressLimitPercent in interface PagingStore

block

public void block()

Specified by:

block in interface PagingStore

unblock

public void unblock()

Specified by:

unblock in interface PagingStore

isRejectingMessages

public boolean isRejectingMessages()

Specified by:

isRejectingMessages in interface PagingStore

getCurrentIds

public Collection<Integer> getCurrentIds()
                                  throws Exception

Description copied from interface: PagingStore

Files to synchronize with a remote backup.

Specified by:

getCurrentIds in interface PagingStore

Returns:

a collection of page IDs which must be synchronized with a replicating backup

Throws:

Exception

sendPages

public void sendPages(ReplicationManager replicator,
 Collection<Integer> pageIds)
               throws Exception

Description copied from interface: PagingStore

Sends the pages with given IDs to the ReplicationManager.

Sending is done here to avoid exposing the internal SequentialFiles.

Specified by:

sendPages in interface PagingStore

Throws:

Exception

getUsedPagesSize

protected int getUsedPagesSize()

forEachUsedPage

protected void forEachUsedPage(Consumer<Page> consumerPage)

getStorageManager

public StorageManager getStorageManager()

Specified by:

getStorageManager in interface PagingStore