Source code

001package gudusoft.gsqlparser;
002
003import java.util.concurrent.*;
004import java.util.concurrent.atomic.AtomicInteger;
005import java.util.Map;
006
007/**
008 * Thread-safe parser pool implementation to replace the singleton pattern.
009 * This pool dramatically improves performance in multi-threaded environments
010 * by eliminating synchronized method bottlenecks.
011 * 
012 * Each database vendor has its own pool of parser instances that can be
013 * borrowed and returned for reuse.
014 */
015public class TParserPool {
016    
017    // Pool configuration
018    private static final int DEFAULT_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;
019    private static final int MAX_POOL_SIZE = 64;
020    private static final long BORROW_TIMEOUT_MS = 5000;
021    
022    // Pools for each database vendor
023    private final ConcurrentHashMap<EDbVendor, BlockingQueue<TGSqlParser>> parserPools;
024    private final ConcurrentHashMap<EDbVendor, AtomicInteger> poolSizes;
025    private final ConcurrentHashMap<EDbVendor, Semaphore> poolSemaphores;
026    
027    // Pool statistics for monitoring
028    private final ConcurrentHashMap<EDbVendor, AtomicInteger> borrowCount;
029    private final ConcurrentHashMap<EDbVendor, AtomicInteger> returnCount;
030    private final ConcurrentHashMap<EDbVendor, AtomicInteger> createCount;
031    
032    private final int poolSize;
033    private volatile boolean shutdown = false;
034    
035    /**
036     * Creates a parser pool with default size
037     */
038    public TParserPool() {
039        this(DEFAULT_POOL_SIZE);
040    }
041    
042    /**
043     * Creates a parser pool with specified size
044     * @param poolSize Size of the pool for each vendor
045     */
046    public TParserPool(int poolSize) {
047        this.poolSize = Math.min(poolSize, MAX_POOL_SIZE);
048        this.parserPools = new ConcurrentHashMap<>();
049        this.poolSizes = new ConcurrentHashMap<>();
050        this.poolSemaphores = new ConcurrentHashMap<>();
051        this.borrowCount = new ConcurrentHashMap<>();
052        this.returnCount = new ConcurrentHashMap<>();
053        this.createCount = new ConcurrentHashMap<>();
054    }
055    
056    /**
057     * Borrows a parser from the pool for the specified vendor.
058     * If no parser is available, creates a new one up to the pool limit.
059     * 
060     * @param vendor Database vendor
061     * @return Parser instance
062     * @throws InterruptedException if interrupted while waiting
063     */
064    public TGSqlParser borrowParser(EDbVendor vendor) throws InterruptedException {
065        if (shutdown) {
066            throw new IllegalStateException("Parser pool has been shut down");
067        }
068        
069        // Get or create the semaphore for this vendor
070        Semaphore semaphore = poolSemaphores.computeIfAbsent(vendor, 
071            v -> new Semaphore(poolSize, true));
072        
073        // Try to acquire a permit with timeout
074        if (!semaphore.tryAcquire(BORROW_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
075            throw new TimeoutException("Timeout waiting for available parser");
076        }
077        
078        try {
079            // Get or create the pool for this vendor
080            BlockingQueue<TGSqlParser> pool = parserPools.computeIfAbsent(vendor, 
081                v -> new LinkedBlockingQueue<>());
082            
083            // Update borrow statistics
084            borrowCount.computeIfAbsent(vendor, v -> new AtomicInteger(0)).incrementAndGet();
085            
086            // Try to get an existing parser from the pool
087            TGSqlParser parser = pool.poll();
088            
089            if (parser == null) {
090                // No parser available, create a new one
091                parser = createParser(vendor);
092                AtomicInteger currentSize = poolSizes.computeIfAbsent(vendor, 
093                    v -> new AtomicInteger(0));
094                currentSize.incrementAndGet();
095                createCount.computeIfAbsent(vendor, v -> new AtomicInteger(0)).incrementAndGet();
096            }
097            
098            // Reset parser state before returning
099            resetParser(parser);
100            return parser;
101            
102        } catch (Exception e) {
103            // Release permit if we fail to get a parser
104            semaphore.release();
105            throw new RuntimeException("Failed to borrow parser", e);
106        }
107    }
108    
109    /**
110     * Returns a parser to the pool for reuse.
111     * 
112     * @param vendor Database vendor
113     * @param parser Parser instance to return
114     */
115    public void returnParser(EDbVendor vendor, TGSqlParser parser) {
116        if (parser == null) {
117            return;
118        }
119        
120        if (shutdown) {
121            // Don't return to pool if shutting down
122            return;
123        }
124        
125        BlockingQueue<TGSqlParser> pool = parserPools.get(vendor);
126        if (pool != null) {
127            // Reset parser state before returning to pool
128            resetParser(parser);
129            
130            // Try to return parser to pool
131            if (pool.offer(parser)) {
132                returnCount.computeIfAbsent(vendor, v -> new AtomicInteger(0)).incrementAndGet();
133            }
134        }
135        
136        // Release the semaphore permit
137        Semaphore semaphore = poolSemaphores.get(vendor);
138        if (semaphore != null) {
139            semaphore.release();
140        }
141    }
142    
143    /**
144     * Executes a function with a borrowed parser and automatically returns it.
145     * This is the recommended way to use the pool.
146     * 
147     * @param vendor Database vendor
148     * @param function Function to execute with the parser
149     * @return Result of the function
150     */
151    public <T> T withParser(EDbVendor vendor, ParserFunction<T> function) throws Exception {
152        TGSqlParser parser = null;
153        try {
154            parser = borrowParser(vendor);
155            return function.apply(parser);
156        } finally {
157            if (parser != null) {
158                returnParser(vendor, parser);
159            }
160        }
161    }
162    
163    /**
164     * Creates a new parser instance for the specified vendor.
165     */
166    private TGSqlParser createParser(EDbVendor vendor) {
167        return new TGSqlParser(vendor);
168    }
169    
170    /**
171     * Resets parser state for reuse.
172     */
173    private void resetParser(TGSqlParser parser) {
174        if (parser != null) {
175            // Clear previous SQL text and results
176            parser.sqltext = null;
177            parser.sqlfilename = null;
178            // And the stream: a pooled parser that kept an exhausted stream from
179            // an earlier borrow served the NEXT borrower zero statements, because
180            // an input stream outranks sqltext when the source is chosen.
181            parser.setSqlInputStream(null);
182            if (parser.sourcetokenlist != null) {
183                parser.sourcetokenlist.clear();
184            }
185            if (parser.sqlstatements != null) {
186                parser.sqlstatements.clear();
187            }
188            // Prevent opt-in SQL Server syntax from leaking to the next borrower.
189            parser.setEnableMssqlColonBindVariables(false);
190            // Note: Some internal parser state cannot be reset from outside
191            // The parser will handle this internally on next parse
192        }
193    }
194    
195    /**
196     * Gets pool statistics for monitoring.
197     */
198    public PoolStatistics getStatistics(EDbVendor vendor) {
199        return new PoolStatistics(
200            poolSizes.getOrDefault(vendor, new AtomicInteger(0)).get(),
201            borrowCount.getOrDefault(vendor, new AtomicInteger(0)).get(),
202            returnCount.getOrDefault(vendor, new AtomicInteger(0)).get(),
203            createCount.getOrDefault(vendor, new AtomicInteger(0)).get(),
204            parserPools.containsKey(vendor) ? parserPools.get(vendor).size() : 0
205        );
206    }
207    
208    /**
209     * Pre-warms the pool by creating parsers in advance.
210     * This avoids the initialization cost during actual usage.
211     * 
212     * @param vendor Database vendor
213     * @param count Number of parsers to pre-create (up to poolSize)
214     */
215    public void prewarm(EDbVendor vendor, int count) {
216        if (shutdown) {
217            throw new IllegalStateException("Parser pool has been shut down");
218        }
219        
220        int toCreate = Math.min(count, poolSize);
221        BlockingQueue<TGSqlParser> pool = parserPools.computeIfAbsent(vendor, 
222            v -> new LinkedBlockingQueue<>());
223        
224        for (int i = 0; i < toCreate; i++) {
225            TGSqlParser parser = createParser(vendor);
226            resetParser(parser);
227            pool.offer(parser);
228            poolSizes.computeIfAbsent(vendor, v -> new AtomicInteger(0)).incrementAndGet();
229            createCount.computeIfAbsent(vendor, v -> new AtomicInteger(0)).incrementAndGet();
230        }
231    }
232    
233    /**
234     * Shuts down the pool and clears all parsers.
235     */
236    public void shutdown() {
237        shutdown = true;
238        parserPools.clear();
239        poolSizes.clear();
240        poolSemaphores.clear();
241    }
242    
243    /**
244     * Functional interface for parser operations.
245     */
246    @FunctionalInterface
247    public interface ParserFunction<T> {
248        T apply(TGSqlParser parser) throws Exception;
249    }
250    
251    /**
252     * Pool statistics for monitoring.
253     */
254    public static class PoolStatistics {
255        public final int totalParsers;
256        public final int borrowCount;
257        public final int returnCount;
258        public final int createCount;
259        public final int availableParsers;
260        
261        public PoolStatistics(int totalParsers, int borrowCount, int returnCount, 
262                             int createCount, int availableParsers) {
263            this.totalParsers = totalParsers;
264            this.borrowCount = borrowCount;
265            this.returnCount = returnCount;
266            this.createCount = createCount;
267            this.availableParsers = availableParsers;
268        }
269        
270        @Override
271        public String toString() {
272            return String.format("PoolStats[total=%d, available=%d, borrows=%d, returns=%d, creates=%d]",
273                totalParsers, availableParsers, borrowCount, returnCount, createCount);
274        }
275    }
276    
277    /**
278     * Custom exception for timeout scenarios.
279     */
280    public static class TimeoutException extends RuntimeException {
281        public TimeoutException(String message) {
282            super(message);
283        }
284    }
285}