Class WriterPool

core/src/main/java/io/questdb/cairo/pool/WriterPool.java:75–714

This class maintains cache of open writers to avoid OS overhead of opening and closing files. While doing so it abides by the same rule as non-pooled writers: there can only be one TableWriter instance for any given table name. This implementation is thread-safe. Writer allocated by one thread c

 * closed.
 */
public class WriterPool extends AbstractPool {
    public static final String OWNERSHIP_REASON_MISSING = "missing or owned by other process";
    public static final String OWNERSHIP_REASON_NONE = null;
    public static final String OWNERSHIP_REASON_RELEASED = "released";
    public static final String OWNERSHIP_REASON_UNKNOWN = "unknown";
    static final String OWNERSHIP_REASON_WRITER_ERROR = "writer error";
    private static final long ENTRY_OWNER = Unsafe.getFieldOffset(Entry.class, "owner");
    private static final Log LOG = LogFactory.getLog(WriterPool.class);
    private static final long QUEUE_PROCESSING_OWNER = -2L;
    private final MicrosecondClock clock;
    private final CairoConfiguration configuration;
    @NotNull
    private final CairoEngine engine;
    private final ConcurrentHashMap<Entry> entries = new ConcurrentHashMap<>();
    private final RecentWriteTracker recentWriteTracker;
    private final CharSequence root;

    /**
     * Pool constructor. WriterPool root directory is passed via configuration.
     *
     * @param configuration      configuration parameters.
     * @param engine             engine instance.
     * @param recentWriteTracker tracker for recent table writes.
     */
    public WriterPool(CairoConfiguration configuration, @NotNull CairoEngine engine, @NotNull RecentWriteTracker recentWriteTracker) {
        super(configuration, configuration.getInactiveWriterTTL());
        this.configuration = configuration;
        this.clock = configuration.getMicrosecondClock();
        this.root = configuration.getDbRoot();
        this.engine = engine;
        this.recentWriteTracker = recentWriteTracker;
        notifyListener(Thread.currentThread().threadId(), null, PoolListener.EV_POOL_OPEN);
    }

    @TestOnly
    public int countFreeWriters() {
        int count = 0;
        for (Entry e : entries.values()) {
            final long owner = e.owner;
            if (owner == UNALLOCATED) {
                count++;
            } else {
                LOG.info().$("table is still busy [table=").$(e.writer.getTableToken())
                        .$(", owner=").$(owner)
                        .I$();
            }
        }
        return count;
    }

    public Map<CharSequence, Entry> entries() {
        return entries;
    }

    /**
     * <p>
     * Creates or retrieves existing TableWriter from pool. Because of TableWriter compliance with <b>single
     * writer model</b> pool ensures there is single TableWriter instance for given table name. Table name is unique in