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, and routine recovery, from leaking to the next borrower.
189            parser.setEnableMssqlColonBindVariables(false);
190            parser.setRecoverRoutineBodyErrors(false);
191            // Note: Some internal parser state cannot be reset from outside
192            // The parser will handle this internally on next parse
193        }
194    }
195    
196    /**
197     * Gets pool statistics for monitoring.
198     */
199    public PoolStatistics getStatistics(EDbVendor vendor) {
200        return new PoolStatistics(
201            poolSizes.getOrDefault(vendor, new AtomicInteger(0)).get(),
202            borrowCount.getOrDefault(vendor, new AtomicInteger(0)).get(),
203            returnCount.getOrDefault(vendor, new AtomicInteger(0)).get(),
204            createCount.getOrDefault(vendor, new AtomicInteger(0)).get(),
205            parserPools.containsKey(vendor) ? parserPools.get(vendor).size() : 0
206        );
207    }
208    
209    /**
210     * Pre-warms the pool by creating parsers in advance.
211     * This avoids the initialization cost during actual usage.
212     * 
213     * @param vendor Database vendor
214     * @param count Number of parsers to pre-create (up to poolSize)
215     */
216    public void prewarm(EDbVendor vendor, int count) {
217        if (shutdown) {
218            throw new IllegalStateException("Parser pool has been shut down");
219        }
220        
221        int toCreate = Math.min(count, poolSize);
222        BlockingQueue<TGSqlParser> pool = parserPools.computeIfAbsent(vendor, 
223            v -> new LinkedBlockingQueue<>());
224        
225        for (int i = 0; i < toCreate; i++) {
226            TGSqlParser parser = createParser(vendor);
227            resetParser(parser);
228            pool.offer(parser);
229            poolSizes.computeIfAbsent(vendor, v -> new AtomicInteger(0)).incrementAndGet();
230            createCount.computeIfAbsent(vendor, v -> new AtomicInteger(0)).incrementAndGet();
231        }
232    }
233    
234    /**
235     * Shuts down the pool and clears all parsers.
236     */
237    public void shutdown() {
238        shutdown = true;
239        parserPools.clear();
240        poolSizes.clear();
241        poolSemaphores.clear();
242    }
243    
244    /**
245     * Functional interface for parser operations.
246     */
247    @FunctionalInterface
248    public interface ParserFunction<T> {
249        T apply(TGSqlParser parser) throws Exception;
250    }
251    
252    /**
253     * Pool statistics for monitoring.
254     */
255    public static class PoolStatistics {
256        public final int totalParsers;
257        public final int borrowCount;
258        public final int returnCount;
259        public final int createCount;
260        public final int availableParsers;
261        
262        public PoolStatistics(int totalParsers, int borrowCount, int returnCount, 
263                             int createCount, int availableParsers) {
264            this.totalParsers = totalParsers;
265            this.borrowCount = borrowCount;
266            this.returnCount = returnCount;
267            this.createCount = createCount;
268            this.availableParsers = availableParsers;
269        }
270        
271        @Override
272        public String toString() {
273            return String.format("PoolStats[total=%d, available=%d, borrows=%d, returns=%d, creates=%d]",
274                totalParsers, availableParsers, borrowCount, returnCount, createCount);
275        }
276    }
277    
278    /**
279     * Custom exception for timeout scenarios.
280     */
281    public static class TimeoutException extends RuntimeException {
282        public TimeoutException(String message) {
283            super(message);
284        }
285    }
286}