http://git-wip-us.apache.org/repos/asf/hbase-site/blob/67e3bccd/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HRegion.RowLockImpl.html ---------------------------------------------------------------------- diff --git a/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HRegion.RowLockImpl.html b/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HRegion.RowLockImpl.html index 6cd6a17..85dd23b 100644 --- a/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HRegion.RowLockImpl.html +++ b/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HRegion.RowLockImpl.html @@ -2119,6526 +2119,6532 @@ <span class="sourceLineNo">2111</span> }<a name="line.2111"></a> <span class="sourceLineNo">2112</span><a name="line.2112"></a> <span class="sourceLineNo">2113</span> status = TaskMonitor.get().createStatus("Compacting " + store + " in " + this);<a name="line.2113"></a> -<span class="sourceLineNo">2114</span> if (this.closed.get()) {<a name="line.2114"></a> -<span class="sourceLineNo">2115</span> String msg = "Skipping compaction on " + this + " because closed";<a name="line.2115"></a> -<span class="sourceLineNo">2116</span> LOG.debug(msg);<a name="line.2116"></a> -<span class="sourceLineNo">2117</span> status.abort(msg);<a name="line.2117"></a> -<span class="sourceLineNo">2118</span> return false;<a name="line.2118"></a> -<span class="sourceLineNo">2119</span> }<a name="line.2119"></a> -<span class="sourceLineNo">2120</span> boolean wasStateSet = false;<a name="line.2120"></a> -<span class="sourceLineNo">2121</span> try {<a name="line.2121"></a> -<span class="sourceLineNo">2122</span> synchronized (writestate) {<a name="line.2122"></a> -<span class="sourceLineNo">2123</span> if (writestate.writesEnabled) {<a name="line.2123"></a> -<span class="sourceLineNo">2124</span> wasStateSet = true;<a name="line.2124"></a> -<span class="sourceLineNo">2125</span> writestate.compacting.incrementAndGet();<a name="line.2125"></a> -<span class="sourceLineNo">2126</span> } else {<a name="line.2126"></a> -<span class="sourceLineNo">2127</span> String msg = "NOT compacting region " + this + ". Writes disabled.";<a name="line.2127"></a> -<span class="sourceLineNo">2128</span> LOG.info(msg);<a name="line.2128"></a> -<span class="sourceLineNo">2129</span> status.abort(msg);<a name="line.2129"></a> -<span class="sourceLineNo">2130</span> return false;<a name="line.2130"></a> -<span class="sourceLineNo">2131</span> }<a name="line.2131"></a> -<span class="sourceLineNo">2132</span> }<a name="line.2132"></a> -<span class="sourceLineNo">2133</span> LOG.info("Starting compaction of {} in {}{}", store, this,<a name="line.2133"></a> -<span class="sourceLineNo">2134</span> (compaction.getRequest().isOffPeak()?" as an off-peak compaction":""));<a name="line.2134"></a> -<span class="sourceLineNo">2135</span> doRegionCompactionPrep();<a name="line.2135"></a> -<span class="sourceLineNo">2136</span> try {<a name="line.2136"></a> -<span class="sourceLineNo">2137</span> status.setStatus("Compacting store " + store);<a name="line.2137"></a> -<span class="sourceLineNo">2138</span> // We no longer need to cancel the request on the way out of this<a name="line.2138"></a> -<span class="sourceLineNo">2139</span> // method because Store#compact will clean up unconditionally<a name="line.2139"></a> -<span class="sourceLineNo">2140</span> requestNeedsCancellation = false;<a name="line.2140"></a> -<span class="sourceLineNo">2141</span> store.compact(compaction, throughputController, user);<a name="line.2141"></a> -<span class="sourceLineNo">2142</span> } catch (InterruptedIOException iioe) {<a name="line.2142"></a> -<span class="sourceLineNo">2143</span> String msg = "compaction interrupted";<a name="line.2143"></a> -<span class="sourceLineNo">2144</span> LOG.info(msg, iioe);<a name="line.2144"></a> -<span class="sourceLineNo">2145</span> status.abort(msg);<a name="line.2145"></a> -<span class="sourceLineNo">2146</span> return false;<a name="line.2146"></a> -<span class="sourceLineNo">2147</span> }<a name="line.2147"></a> -<span class="sourceLineNo">2148</span> } finally {<a name="line.2148"></a> -<span class="sourceLineNo">2149</span> if (wasStateSet) {<a name="line.2149"></a> -<span class="sourceLineNo">2150</span> synchronized (writestate) {<a name="line.2150"></a> -<span class="sourceLineNo">2151</span> writestate.compacting.decrementAndGet();<a name="line.2151"></a> -<span class="sourceLineNo">2152</span> if (writestate.compacting.get() <= 0) {<a name="line.2152"></a> -<span class="sourceLineNo">2153</span> writestate.notifyAll();<a name="line.2153"></a> -<span class="sourceLineNo">2154</span> }<a name="line.2154"></a> -<span class="sourceLineNo">2155</span> }<a name="line.2155"></a> -<span class="sourceLineNo">2156</span> }<a name="line.2156"></a> -<span class="sourceLineNo">2157</span> }<a name="line.2157"></a> -<span class="sourceLineNo">2158</span> status.markComplete("Compaction complete");<a name="line.2158"></a> -<span class="sourceLineNo">2159</span> return true;<a name="line.2159"></a> -<span class="sourceLineNo">2160</span> } finally {<a name="line.2160"></a> -<span class="sourceLineNo">2161</span> if (requestNeedsCancellation) store.cancelRequestedCompaction(compaction);<a name="line.2161"></a> -<span class="sourceLineNo">2162</span> if (status != null) status.cleanup();<a name="line.2162"></a> -<span class="sourceLineNo">2163</span> }<a name="line.2163"></a> -<span class="sourceLineNo">2164</span> }<a name="line.2164"></a> -<span class="sourceLineNo">2165</span><a name="line.2165"></a> -<span class="sourceLineNo">2166</span> /**<a name="line.2166"></a> -<span class="sourceLineNo">2167</span> * Flush the cache.<a name="line.2167"></a> -<span class="sourceLineNo">2168</span> *<a name="line.2168"></a> -<span class="sourceLineNo">2169</span> * <p>When this method is called the cache will be flushed unless:<a name="line.2169"></a> -<span class="sourceLineNo">2170</span> * <ol><a name="line.2170"></a> -<span class="sourceLineNo">2171</span> * <li>the cache is empty</li><a name="line.2171"></a> -<span class="sourceLineNo">2172</span> * <li>the region is closed.</li><a name="line.2172"></a> -<span class="sourceLineNo">2173</span> * <li>a flush is already in progress</li><a name="line.2173"></a> -<span class="sourceLineNo">2174</span> * <li>writes are disabled</li><a name="line.2174"></a> -<span class="sourceLineNo">2175</span> * </ol><a name="line.2175"></a> -<span class="sourceLineNo">2176</span> *<a name="line.2176"></a> -<span class="sourceLineNo">2177</span> * <p>This method may block for some time, so it should not be called from a<a name="line.2177"></a> -<span class="sourceLineNo">2178</span> * time-sensitive thread.<a name="line.2178"></a> -<span class="sourceLineNo">2179</span> * @param force whether we want to force a flush of all stores<a name="line.2179"></a> -<span class="sourceLineNo">2180</span> * @return FlushResult indicating whether the flush was successful or not and if<a name="line.2180"></a> -<span class="sourceLineNo">2181</span> * the region needs compacting<a name="line.2181"></a> -<span class="sourceLineNo">2182</span> *<a name="line.2182"></a> -<span class="sourceLineNo">2183</span> * @throws IOException general io exceptions<a name="line.2183"></a> -<span class="sourceLineNo">2184</span> * because a snapshot was not properly persisted.<a name="line.2184"></a> -<span class="sourceLineNo">2185</span> */<a name="line.2185"></a> -<span class="sourceLineNo">2186</span> // TODO HBASE-18905. We might have to expose a requestFlush API for CPs<a name="line.2186"></a> -<span class="sourceLineNo">2187</span> public FlushResult flush(boolean force) throws IOException {<a name="line.2187"></a> -<span class="sourceLineNo">2188</span> return flushcache(force, false, FlushLifeCycleTracker.DUMMY);<a name="line.2188"></a> -<span class="sourceLineNo">2189</span> }<a name="line.2189"></a> -<span class="sourceLineNo">2190</span><a name="line.2190"></a> -<span class="sourceLineNo">2191</span> public interface FlushResult {<a name="line.2191"></a> -<span class="sourceLineNo">2192</span> enum Result {<a name="line.2192"></a> -<span class="sourceLineNo">2193</span> FLUSHED_NO_COMPACTION_NEEDED,<a name="line.2193"></a> -<span class="sourceLineNo">2194</span> FLUSHED_COMPACTION_NEEDED,<a name="line.2194"></a> -<span class="sourceLineNo">2195</span> // Special case where a flush didn't run because there's nothing in the memstores. Used when<a name="line.2195"></a> -<span class="sourceLineNo">2196</span> // bulk loading to know when we can still load even if a flush didn't happen.<a name="line.2196"></a> -<span class="sourceLineNo">2197</span> CANNOT_FLUSH_MEMSTORE_EMPTY,<a name="line.2197"></a> -<span class="sourceLineNo">2198</span> CANNOT_FLUSH<a name="line.2198"></a> -<span class="sourceLineNo">2199</span> }<a name="line.2199"></a> -<span class="sourceLineNo">2200</span><a name="line.2200"></a> -<span class="sourceLineNo">2201</span> /** @return the detailed result code */<a name="line.2201"></a> -<span class="sourceLineNo">2202</span> Result getResult();<a name="line.2202"></a> -<span class="sourceLineNo">2203</span><a name="line.2203"></a> -<span class="sourceLineNo">2204</span> /** @return true if the memstores were flushed, else false */<a name="line.2204"></a> -<span class="sourceLineNo">2205</span> boolean isFlushSucceeded();<a name="line.2205"></a> -<span class="sourceLineNo">2206</span><a name="line.2206"></a> -<span class="sourceLineNo">2207</span> /** @return True if the flush requested a compaction, else false */<a name="line.2207"></a> -<span class="sourceLineNo">2208</span> boolean isCompactionNeeded();<a name="line.2208"></a> -<span class="sourceLineNo">2209</span> }<a name="line.2209"></a> +<span class="sourceLineNo">2114</span> status.enableStatusJournal(false);<a name="line.2114"></a> +<span class="sourceLineNo">2115</span> if (this.closed.get()) {<a name="line.2115"></a> +<span class="sourceLineNo">2116</span> String msg = "Skipping compaction on " + this + " because closed";<a name="line.2116"></a> +<span class="sourceLineNo">2117</span> LOG.debug(msg);<a name="line.2117"></a> +<span class="sourceLineNo">2118</span> status.abort(msg);<a name="line.2118"></a> +<span class="sourceLineNo">2119</span> return false;<a name="line.2119"></a> +<span class="sourceLineNo">2120</span> }<a name="line.2120"></a> +<span class="sourceLineNo">2121</span> boolean wasStateSet = false;<a name="line.2121"></a> +<span class="sourceLineNo">2122</span> try {<a name="line.2122"></a> +<span class="sourceLineNo">2123</span> synchronized (writestate) {<a name="line.2123"></a> +<span class="sourceLineNo">2124</span> if (writestate.writesEnabled) {<a name="line.2124"></a> +<span class="sourceLineNo">2125</span> wasStateSet = true;<a name="line.2125"></a> +<span class="sourceLineNo">2126</span> writestate.compacting.incrementAndGet();<a name="line.2126"></a> +<span class="sourceLineNo">2127</span> } else {<a name="line.2127"></a> +<span class="sourceLineNo">2128</span> String msg = "NOT compacting region " + this + ". Writes disabled.";<a name="line.2128"></a> +<span class="sourceLineNo">2129</span> LOG.info(msg);<a name="line.2129"></a> +<span class="sourceLineNo">2130</span> status.abort(msg);<a name="line.2130"></a> +<span class="sourceLineNo">2131</span> return false;<a name="line.2131"></a> +<span class="sourceLineNo">2132</span> }<a name="line.2132"></a> +<span class="sourceLineNo">2133</span> }<a name="line.2133"></a> +<span class="sourceLineNo">2134</span> LOG.info("Starting compaction of {} in {}{}", store, this,<a name="line.2134"></a> +<span class="sourceLineNo">2135</span> (compaction.getRequest().isOffPeak()?" as an off-peak compaction":""));<a name="line.2135"></a> +<span class="sourceLineNo">2136</span> doRegionCompactionPrep();<a name="line.2136"></a> +<span class="sourceLineNo">2137</span> try {<a name="line.2137"></a> +<span class="sourceLineNo">2138</span> status.setStatus("Compacting store " + store);<a name="line.2138"></a> +<span class="sourceLineNo">2139</span> // We no longer need to cancel the request on the way out of this<a name="line.2139"></a> +<span class="sourceLineNo">2140</span> // method because Store#compact will clean up unconditionally<a name="line.2140"></a> +<span class="sourceLineNo">2141</span> requestNeedsCancellation = false;<a name="line.2141"></a> +<span class="sourceLineNo">2142</span> store.compact(compaction, throughputController, user);<a name="line.2142"></a> +<span class="sourceLineNo">2143</span> } catch (InterruptedIOException iioe) {<a name="line.2143"></a> +<span class="sourceLineNo">2144</span> String msg = "compaction interrupted";<a name="line.2144"></a> +<span class="sourceLineNo">2145</span> LOG.info(msg, iioe);<a name="line.2145"></a> +<span class="sourceLineNo">2146</span> status.abort(msg);<a name="line.2146"></a> +<span class="sourceLineNo">2147</span> return false;<a name="line.2147"></a> +<span class="sourceLineNo">2148</span> }<a name="line.2148"></a> +<span class="sourceLineNo">2149</span> } finally {<a name="line.2149"></a> +<span class="sourceLineNo">2150</span> if (wasStateSet) {<a name="line.2150"></a> +<span class="sourceLineNo">2151</span> synchronized (writestate) {<a name="line.2151"></a> +<span class="sourceLineNo">2152</span> writestate.compacting.decrementAndGet();<a name="line.2152"></a> +<span class="sourceLineNo">2153</span> if (writestate.compacting.get() <= 0) {<a name="line.2153"></a> +<span class="sourceLineNo">2154</span> writestate.notifyAll();<a name="line.2154"></a> +<span class="sourceLineNo">2155</span> }<a name="line.2155"></a> +<span class="sourceLineNo">2156</span> }<a name="line.2156"></a> +<span class="sourceLineNo">2157</span> }<a name="line.2157"></a> +<span class="sourceLineNo">2158</span> }<a name="line.2158"></a> +<span class="sourceLineNo">2159</span> status.markComplete("Compaction complete");<a name="line.2159"></a> +<span class="sourceLineNo">2160</span> return true;<a name="line.2160"></a> +<span class="sourceLineNo">2161</span> } finally {<a name="line.2161"></a> +<span class="sourceLineNo">2162</span> if (requestNeedsCancellation) store.cancelRequestedCompaction(compaction);<a name="line.2162"></a> +<span class="sourceLineNo">2163</span> if (status != null) {<a name="line.2163"></a> +<span class="sourceLineNo">2164</span> LOG.debug("Compaction status journal:\n\t" + status.prettyPrintJournal());<a name="line.2164"></a> +<span class="sourceLineNo">2165</span> status.cleanup();<a name="line.2165"></a> +<span class="sourceLineNo">2166</span> }<a name="line.2166"></a> +<span class="sourceLineNo">2167</span> }<a name="line.2167"></a> +<span class="sourceLineNo">2168</span> }<a name="line.2168"></a> +<span class="sourceLineNo">2169</span><a name="line.2169"></a> +<span class="sourceLineNo">2170</span> /**<a name="line.2170"></a> +<span class="sourceLineNo">2171</span> * Flush the cache.<a name="line.2171"></a> +<span class="sourceLineNo">2172</span> *<a name="line.2172"></a> +<span class="sourceLineNo">2173</span> * <p>When this method is called the cache will be flushed unless:<a name="line.2173"></a> +<span class="sourceLineNo">2174</span> * <ol><a name="line.2174"></a> +<span class="sourceLineNo">2175</span> * <li>the cache is empty</li><a name="line.2175"></a> +<span class="sourceLineNo">2176</span> * <li>the region is closed.</li><a name="line.2176"></a> +<span class="sourceLineNo">2177</span> * <li>a flush is already in progress</li><a name="line.2177"></a> +<span class="sourceLineNo">2178</span> * <li>writes are disabled</li><a name="line.2178"></a> +<span class="sourceLineNo">2179</span> * </ol><a name="line.2179"></a> +<span class="sourceLineNo">2180</span> *<a name="line.2180"></a> +<span class="sourceLineNo">2181</span> * <p>This method may block for some time, so it should not be called from a<a name="line.2181"></a> +<span class="sourceLineNo">2182</span> * time-sensitive thread.<a name="line.2182"></a> +<span class="sourceLineNo">2183</span> * @param force whether we want to force a flush of all stores<a name="line.2183"></a> +<span class="sourceLineNo">2184</span> * @return FlushResult indicating whether the flush was successful or not and if<a name="line.2184"></a> +<span class="sourceLineNo">2185</span> * the region needs compacting<a name="line.2185"></a> +<span class="sourceLineNo">2186</span> *<a name="line.2186"></a> +<span class="sourceLineNo">2187</span> * @throws IOException general io exceptions<a name="line.2187"></a> +<span class="sourceLineNo">2188</span> * because a snapshot was not properly persisted.<a name="line.2188"></a> +<span class="sourceLineNo">2189</span> */<a name="line.2189"></a> +<span class="sourceLineNo">2190</span> // TODO HBASE-18905. We might have to expose a requestFlush API for CPs<a name="line.2190"></a> +<span class="sourceLineNo">2191</span> public FlushResult flush(boolean force) throws IOException {<a name="line.2191"></a> +<span class="sourceLineNo">2192</span> return flushcache(force, false, FlushLifeCycleTracker.DUMMY);<a name="line.2192"></a> +<span class="sourceLineNo">2193</span> }<a name="line.2193"></a> +<span class="sourceLineNo">2194</span><a name="line.2194"></a> +<span class="sourceLineNo">2195</span> public interface FlushResult {<a name="line.2195"></a> +<span class="sourceLineNo">2196</span> enum Result {<a name="line.2196"></a> +<span class="sourceLineNo">2197</span> FLUSHED_NO_COMPACTION_NEEDED,<a name="line.2197"></a> +<span class="sourceLineNo">2198</span> FLUSHED_COMPACTION_NEEDED,<a name="line.2198"></a> +<span class="sourceLineNo">2199</span> // Special case where a flush didn't run because there's nothing in the memstores. Used when<a name="line.2199"></a> +<span class="sourceLineNo">2200</span> // bulk loading to know when we can still load even if a flush didn't happen.<a name="line.2200"></a> +<span class="sourceLineNo">2201</span> CANNOT_FLUSH_MEMSTORE_EMPTY,<a name="line.2201"></a> +<span class="sourceLineNo">2202</span> CANNOT_FLUSH<a name="line.2202"></a> +<span class="sourceLineNo">2203</span> }<a name="line.2203"></a> +<span class="sourceLineNo">2204</span><a name="line.2204"></a> +<span class="sourceLineNo">2205</span> /** @return the detailed result code */<a name="line.2205"></a> +<span class="sourceLineNo">2206</span> Result getResult();<a name="line.2206"></a> +<span class="sourceLineNo">2207</span><a name="line.2207"></a> +<span class="sourceLineNo">2208</span> /** @return true if the memstores were flushed, else false */<a name="line.2208"></a> +<span class="sourceLineNo">2209</span> boolean isFlushSucceeded();<a name="line.2209"></a> <span class="sourceLineNo">2210</span><a name="line.2210"></a> -<span class="sourceLineNo">2211</span> /**<a name="line.2211"></a> -<span class="sourceLineNo">2212</span> * Flush the cache.<a name="line.2212"></a> -<span class="sourceLineNo">2213</span> *<a name="line.2213"></a> -<span class="sourceLineNo">2214</span> * When this method is called the cache will be flushed unless:<a name="line.2214"></a> -<span class="sourceLineNo">2215</span> * <ol><a name="line.2215"></a> -<span class="sourceLineNo">2216</span> * <li>the cache is empty</li><a name="line.2216"></a> -<span class="sourceLineNo">2217</span> * <li>the region is closed.</li><a name="line.2217"></a> -<span class="sourceLineNo">2218</span> * <li>a flush is already in progress</li><a name="line.2218"></a> -<span class="sourceLineNo">2219</span> * <li>writes are disabled</li><a name="line.2219"></a> -<span class="sourceLineNo">2220</span> * </ol><a name="line.2220"></a> -<span class="sourceLineNo">2221</span> *<a name="line.2221"></a> -<span class="sourceLineNo">2222</span> * <p>This method may block for some time, so it should not be called from a<a name="line.2222"></a> -<span class="sourceLineNo">2223</span> * time-sensitive thread.<a name="line.2223"></a> -<span class="sourceLineNo">2224</span> * @param forceFlushAllStores whether we want to flush all stores<a name="line.2224"></a> -<span class="sourceLineNo">2225</span> * @param writeFlushRequestWalMarker whether to write the flush request marker to WAL<a name="line.2225"></a> -<span class="sourceLineNo">2226</span> * @param tracker used to track the life cycle of this flush<a name="line.2226"></a> -<span class="sourceLineNo">2227</span> * @return whether the flush is success and whether the region needs compacting<a name="line.2227"></a> -<span class="sourceLineNo">2228</span> *<a name="line.2228"></a> -<span class="sourceLineNo">2229</span> * @throws IOException general io exceptions<a name="line.2229"></a> -<span class="sourceLineNo">2230</span> * @throws DroppedSnapshotException Thrown when replay of wal is required<a name="line.2230"></a> -<span class="sourceLineNo">2231</span> * because a Snapshot was not properly persisted. The region is put in closing mode, and the<a name="line.2231"></a> -<span class="sourceLineNo">2232</span> * caller MUST abort after this.<a name="line.2232"></a> -<span class="sourceLineNo">2233</span> */<a name="line.2233"></a> -<span class="sourceLineNo">2234</span> public FlushResultImpl flushcache(boolean forceFlushAllStores, boolean writeFlushRequestWalMarker,<a name="line.2234"></a> -<span class="sourceLineNo">2235</span> FlushLifeCycleTracker tracker) throws IOException {<a name="line.2235"></a> -<span class="sourceLineNo">2236</span> // fail-fast instead of waiting on the lock<a name="line.2236"></a> -<span class="sourceLineNo">2237</span> if (this.closing.get()) {<a name="line.2237"></a> -<span class="sourceLineNo">2238</span> String msg = "Skipping flush on " + this + " because closing";<a name="line.2238"></a> -<span class="sourceLineNo">2239</span> LOG.debug(msg);<a name="line.2239"></a> -<span class="sourceLineNo">2240</span> return new FlushResultImpl(FlushResult.Result.CANNOT_FLUSH, msg, false);<a name="line.2240"></a> -<span class="sourceLineNo">2241</span> }<a name="line.2241"></a> -<span class="sourceLineNo">2242</span> MonitoredTask status = TaskMonitor.get().createStatus("Flushing " + this);<a name="line.2242"></a> -<span class="sourceLineNo">2243</span> status.setStatus("Acquiring readlock on region");<a name="line.2243"></a> -<span class="sourceLineNo">2244</span> // block waiting for the lock for flushing cache<a name="line.2244"></a> -<span class="sourceLineNo">2245</span> lock.readLock().lock();<a name="line.2245"></a> -<span class="sourceLineNo">2246</span> try {<a name="line.2246"></a> -<span class="sourceLineNo">2247</span> if (this.closed.get()) {<a name="line.2247"></a> -<span class="sourceLineNo">2248</span> String msg = "Skipping flush on " + this + " because closed";<a name="line.2248"></a> -<span class="sourceLineNo">2249</span> LOG.debug(msg);<a name="line.2249"></a> -<span class="sourceLineNo">2250</span> status.abort(msg);<a name="line.2250"></a> -<span class="sourceLineNo">2251</span> return new FlushResultImpl(FlushResult.Result.CANNOT_FLUSH, msg, false);<a name="line.2251"></a> -<span class="sourceLineNo">2252</span> }<a name="line.2252"></a> -<span class="sourceLineNo">2253</span> if (coprocessorHost != null) {<a name="line.2253"></a> -<span class="sourceLineNo">2254</span> status.setStatus("Running coprocessor pre-flush hooks");<a name="line.2254"></a> -<span class="sourceLineNo">2255</span> coprocessorHost.preFlush(tracker);<a name="line.2255"></a> -<span class="sourceLineNo">2256</span> }<a name="line.2256"></a> -<span class="sourceLineNo">2257</span> // TODO: this should be managed within memstore with the snapshot, updated only after flush<a name="line.2257"></a> -<span class="sourceLineNo">2258</span> // successful<a name="line.2258"></a> -<span class="sourceLineNo">2259</span> if (numMutationsWithoutWAL.sum() > 0) {<a name="line.2259"></a> -<span class="sourceLineNo">2260</span> numMutationsWithoutWAL.reset();<a name="line.2260"></a> -<span class="sourceLineNo">2261</span> dataInMemoryWithoutWAL.reset();<a name="line.2261"></a> -<span class="sourceLineNo">2262</span> }<a name="line.2262"></a> -<span class="sourceLineNo">2263</span> synchronized (writestate) {<a name="line.2263"></a> -<span class="sourceLineNo">2264</span> if (!writestate.flushing && writestate.writesEnabled) {<a name="line.2264"></a> -<span class="sourceLineNo">2265</span> this.writestate.flushing = true;<a name="line.2265"></a> -<span class="sourceLineNo">2266</span> } else {<a name="line.2266"></a> -<span class="sourceLineNo">2267</span> if (LOG.isDebugEnabled()) {<a name="line.2267"></a> -<span class="sourceLineNo">2268</span> LOG.debug("NOT flushing memstore for region " + this<a name="line.2268"></a> -<span class="sourceLineNo">2269</span> + ", flushing=" + writestate.flushing + ", writesEnabled="<a name="line.2269"></a> -<span class="sourceLineNo">2270</span> + writestate.writesEnabled);<a name="line.2270"></a> -<span class="sourceLineNo">2271</span> }<a name="line.2271"></a> -<span class="sourceLineNo">2272</span> String msg = "Not flushing since "<a name="line.2272"></a> -<span class="sourceLineNo">2273</span> + (writestate.flushing ? "already flushing"<a name="line.2273"></a> -<span class="sourceLineNo">2274</span> : "writes not enabled");<a name="line.2274"></a> -<span class="sourceLineNo">2275</span> status.abort(msg);<a name="line.2275"></a> -<span class="sourceLineNo">2276</span> return new FlushResultImpl(FlushResult.Result.CANNOT_FLUSH, msg, false);<a name="line.2276"></a> -<span class="sourceLineNo">2277</span> }<a name="line.2277"></a> -<span class="sourceLineNo">2278</span> }<a name="line.2278"></a> -<span class="sourceLineNo">2279</span><a name="line.2279"></a> -<span class="sourceLineNo">2280</span> try {<a name="line.2280"></a> -<span class="sourceLineNo">2281</span> Collection<HStore> specificStoresToFlush =<a name="line.2281"></a> -<span class="sourceLineNo">2282</span> forceFlushAllStores ? stores.values() : flushPolicy.selectStoresToFlush();<a name="line.2282"></a> -<span class="sourceLineNo">2283</span> FlushResultImpl fs =<a name="line.2283"></a> -<span class="sourceLineNo">2284</span> internalFlushcache(specificStoresToFlush, status, writeFlushRequestWalMarker, tracker);<a name="line.2284"></a> -<span class="sourceLineNo">2285</span><a name="line.2285"></a> -<span class="sourceLineNo">2286</span> if (coprocessorHost != null) {<a name="line.2286"></a> -<span class="sourceLineNo">2287</span> status.setStatus("Running post-flush coprocessor hooks");<a name="line.2287"></a> -<span class="sourceLineNo">2288</span> coprocessorHost.postFlush(tracker);<a name="line.2288"></a> -<span class="sourceLineNo">2289</span> }<a name="line.2289"></a> +<span class="sourceLineNo">2211</span> /** @return True if the flush requested a compaction, else false */<a name="line.2211"></a> +<span class="sourceLineNo">2212</span> boolean isCompactionNeeded();<a name="line.2212"></a> +<span class="sourceLineNo">2213</span> }<a name="line.2213"></a> +<span class="sourceLineNo">2214</span><a name="line.2214"></a> +<span class="sourceLineNo">2215</span> /**<a name="line.2215"></a> +<span class="sourceLineNo">2216</span> * Flush the cache.<a name="line.2216"></a> +<span class="sourceLineNo">2217</span> *<a name="line.2217"></a> +<span class="sourceLineNo">2218</span> * When this method is called the cache will be flushed unless:<a name="line.2218"></a> +<span class="sourceLineNo">2219</span> * <ol><a name="line.2219"></a> +<span class="sourceLineNo">2220</span> * <li>the cache is empty</li><a name="line.2220"></a> +<span class="sourceLineNo">2221</span> * <li>the region is closed.</li><a name="line.2221"></a> +<span class="sourceLineNo">2222</span> * <li>a flush is already in progress</li><a name="line.2222"></a> +<span class="sourceLineNo">2223</span> * <li>writes are disabled</li><a name="line.2223"></a> +<span class="sourceLineNo">2224</span> * </ol><a name="line.2224"></a> +<span class="sourceLineNo">2225</span> *<a name="line.2225"></a> +<span class="sourceLineNo">2226</span> * <p>This method may block for some time, so it should not be called from a<a name="line.2226"></a> +<span class="sourceLineNo">2227</span> * time-sensitive thread.<a name="line.2227"></a> +<span class="sourceLineNo">2228</span> * @param forceFlushAllStores whether we want to flush all stores<a name="line.2228"></a> +<span class="sourceLineNo">2229</span> * @param writeFlushRequestWalMarker whether to write the flush request marker to WAL<a name="line.2229"></a> +<span class="sourceLineNo">2230</span> * @param tracker used to track the life cycle of this flush<a name="line.2230"></a> +<span class="sourceLineNo">2231</span> * @return whether the flush is success and whether the region needs compacting<a name="line.2231"></a> +<span class="sourceLineNo">2232</span> *<a name="line.2232"></a> +<span class="sourceLineNo">2233</span> * @throws IOException general io exceptions<a name="line.2233"></a> +<span class="sourceLineNo">2234</span> * @throws DroppedSnapshotException Thrown when replay of wal is required<a name="line.2234"></a> +<span class="sourceLineNo">2235</span> * because a Snapshot was not properly persisted. The region is put in closing mode, and the<a name="line.2235"></a> +<span class="sourceLineNo">2236</span> * caller MUST abort after this.<a name="line.2236"></a> +<span class="sourceLineNo">2237</span> */<a name="line.2237"></a> +<span class="sourceLineNo">2238</span> public FlushResultImpl flushcache(boolean forceFlushAllStores, boolean writeFlushRequestWalMarker,<a name="line.2238"></a> +<span class="sourceLineNo">2239</span> FlushLifeCycleTracker tracker) throws IOException {<a name="line.2239"></a> +<span class="sourceLineNo">2240</span> // fail-fast instead of waiting on the lock<a name="line.2240"></a> +<span class="sourceLineNo">2241</span> if (this.closing.get()) {<a name="line.2241"></a> +<span class="sourceLineNo">2242</span> String msg = "Skipping flush on " + this + " because closing";<a name="line.2242"></a> +<span class="sourceLineNo">2243</span> LOG.debug(msg);<a name="line.2243"></a> +<span class="sourceLineNo">2244</span> return new FlushResultImpl(FlushResult.Result.CANNOT_FLUSH, msg, false);<a name="line.2244"></a> +<span class="sourceLineNo">2245</span> }<a name="line.2245"></a> +<span class="sourceLineNo">2246</span> MonitoredTask status = TaskMonitor.get().createStatus("Flushing " + this);<a name="line.2246"></a> +<span class="sourceLineNo">2247</span> status.enableStatusJournal(false);<a name="line.2247"></a> +<span class="sourceLineNo">2248</span> status.setStatus("Acquiring readlock on region");<a name="line.2248"></a> +<span class="sourceLineNo">2249</span> // block waiting for the lock for flushing cache<a name="line.2249"></a> +<span class="sourceLineNo">2250</span> lock.readLock().lock();<a name="line.2250"></a> +<span class="sourceLineNo">2251</span> try {<a name="line.2251"></a> +<span class="sourceLineNo">2252</span> if (this.closed.get()) {<a name="line.2252"></a> +<span class="sourceLineNo">2253</span> String msg = "Skipping flush on " + this + " because closed";<a name="line.2253"></a> +<span class="sourceLineNo">2254</span> LOG.debug(msg);<a name="line.2254"></a> +<span class="sourceLineNo">2255</span> status.abort(msg);<a name="line.2255"></a> +<span class="sourceLineNo">2256</span> return new FlushResultImpl(FlushResult.Result.CANNOT_FLUSH, msg, false);<a name="line.2256"></a> +<span class="sourceLineNo">2257</span> }<a name="line.2257"></a> +<span class="sourceLineNo">2258</span> if (coprocessorHost != null) {<a name="line.2258"></a> +<span class="sourceLineNo">2259</span> status.setStatus("Running coprocessor pre-flush hooks");<a name="line.2259"></a> +<span class="sourceLineNo">2260</span> coprocessorHost.preFlush(tracker);<a name="line.2260"></a> +<span class="sourceLineNo">2261</span> }<a name="line.2261"></a> +<span class="sourceLineNo">2262</span> // TODO: this should be managed within memstore with the snapshot, updated only after flush<a name="line.2262"></a> +<span class="sourceLineNo">2263</span> // successful<a name="line.2263"></a> +<span class="sourceLineNo">2264</span> if (numMutationsWithoutWAL.sum() > 0) {<a name="line.2264"></a> +<span class="sourceLineNo">2265</span> numMutationsWithoutWAL.reset();<a name="line.2265"></a> +<span class="sourceLineNo">2266</span> dataInMemoryWithoutWAL.reset();<a name="line.2266"></a> +<span class="sourceLineNo">2267</span> }<a name="line.2267"></a> +<span class="sourceLineNo">2268</span> synchronized (writestate) {<a name="line.2268"></a> +<span class="sourceLineNo">2269</span> if (!writestate.flushing && writestate.writesEnabled) {<a name="line.2269"></a> +<span class="sourceLineNo">2270</span> this.writestate.flushing = true;<a name="line.2270"></a> +<span class="sourceLineNo">2271</span> } else {<a name="line.2271"></a> +<span class="sourceLineNo">2272</span> if (LOG.isDebugEnabled()) {<a name="line.2272"></a> +<span class="sourceLineNo">2273</span> LOG.debug("NOT flushing memstore for region " + this<a name="line.2273"></a> +<span class="sourceLineNo">2274</span> + ", flushing=" + writestate.flushing + ", writesEnabled="<a name="line.2274"></a> +<span class="sourceLineNo">2275</span> + writestate.writesEnabled);<a name="line.2275"></a> +<span class="sourceLineNo">2276</span> }<a name="line.2276"></a> +<span class="sourceLineNo">2277</span> String msg = "Not flushing since "<a name="line.2277"></a> +<span class="sourceLineNo">2278</span> + (writestate.flushing ? "already flushing"<a name="line.2278"></a> +<span class="sourceLineNo">2279</span> : "writes not enabled");<a name="line.2279"></a> +<span class="sourceLineNo">2280</span> status.abort(msg);<a name="line.2280"></a> +<span class="sourceLineNo">2281</span> return new FlushResultImpl(FlushResult.Result.CANNOT_FLUSH, msg, false);<a name="line.2281"></a> +<span class="sourceLineNo">2282</span> }<a name="line.2282"></a> +<span class="sourceLineNo">2283</span> }<a name="line.2283"></a> +<span class="sourceLineNo">2284</span><a name="line.2284"></a> +<span class="sourceLineNo">2285</span> try {<a name="line.2285"></a> +<span class="sourceLineNo">2286</span> Collection<HStore> specificStoresToFlush =<a name="line.2286"></a> +<span class="sourceLineNo">2287</span> forceFlushAllStores ? stores.values() : flushPolicy.selectStoresToFlush();<a name="line.2287"></a> +<span class="sourceLineNo">2288</span> FlushResultImpl fs =<a name="line.2288"></a> +<span class="sourceLineNo">2289</span> internalFlushcache(specificStoresToFlush, status, writeFlushRequestWalMarker, tracker);<a name="line.2289"></a> <span class="sourceLineNo">2290</span><a name="line.2290"></a> -<span class="sourceLineNo">2291</span> if(fs.isFlushSucceeded()) {<a name="line.2291"></a> -<span class="sourceLineNo">2292</span> flushesQueued.reset();<a name="line.2292"></a> -<span class="sourceLineNo">2293</span> }<a name="line.2293"></a> -<span class="sourceLineNo">2294</span><a name="line.2294"></a> -<span class="sourceLineNo">2295</span> status.markComplete("Flush successful");<a name="line.2295"></a> -<span class="sourceLineNo">2296</span> return fs;<a name="line.2296"></a> -<span class="sourceLineNo">2297</span> } finally {<a name="line.2297"></a> -<span class="sourceLineNo">2298</span> synchronized (writestate) {<a name="line.2298"></a> -<span class="sourceLineNo">2299</span> writestate.flushing = false;<a name="line.2299"></a> -<span class="sourceLineNo">2300</span> this.writestate.flushRequested = false;<a name="line.2300"></a> -<span class="sourceLineNo">2301</span> writestate.notifyAll();<a name="line.2301"></a> -<span class="sourceLineNo">2302</span> }<a name="line.2302"></a> -<span class="sourceLineNo">2303</span> }<a name="line.2303"></a> -<span class="sourceLineNo">2304</span> } finally {<a name="line.2304"></a> -<span class="sourceLineNo">2305</span> lock.readLock().unlock();<a name="line.2305"></a> -<span class="sourceLineNo">2306</span> status.cleanup();<a name="line.2306"></a> -<span class="sourceLineNo">2307</span> }<a name="line.2307"></a> -<span class="sourceLineNo">2308</span> }<a name="line.2308"></a> -<span class="sourceLineNo">2309</span><a name="line.2309"></a> -<span class="sourceLineNo">2310</span> /**<a name="line.2310"></a> -<span class="sourceLineNo">2311</span> * Should the store be flushed because it is old enough.<a name="line.2311"></a> -<span class="sourceLineNo">2312</span> * <p><a name="line.2312"></a> -<span class="sourceLineNo">2313</span> * Every FlushPolicy should call this to determine whether a store is old enough to flush (except<a name="line.2313"></a> -<span class="sourceLineNo">2314</span> * that you always flush all stores). Otherwise the method will always<a name="line.2314"></a> -<span class="sourceLineNo">2315</span> * returns true which will make a lot of flush requests.<a name="line.2315"></a> -<span class="sourceLineNo">2316</span> */<a name="line.2316"></a> -<span class="sourceLineNo">2317</span> boolean shouldFlushStore(HStore store) {<a name="line.2317"></a> -<span class="sourceLineNo">2318</span> long earliest = this.wal.getEarliestMemStoreSeqNum(getRegionInfo().getEncodedNameAsBytes(),<a name="line.2318"></a> -<span class="sourceLineNo">2319</span> store.getColumnFamilyDescriptor().getName()) - 1;<a name="line.2319"></a> -<span class="sourceLineNo">2320</span> if (earliest > 0 && earliest + flushPerChanges < mvcc.getReadPoint()) {<a name="line.2320"></a> -<span class="sourceLineNo">2321</span> if (LOG.isDebugEnabled()) {<a name="line.2321"></a> -<span class="sourceLineNo">2322</span> LOG.debug("Flush column family " + store.getColumnFamilyName() + " of " +<a name="line.2322"></a> -<span class="sourceLineNo">2323</span> getRegionInfo().getEncodedName() + " because unflushed sequenceid=" + earliest +<a name="line.2323"></a> -<span class="sourceLineNo">2324</span> " is > " + this.flushPerChanges + " from current=" + mvcc.getReadPoint());<a name="line.2324"></a> -<span class="sourceLineNo">2325</span> }<a name="line.2325"></a> -<span class="sourceLineNo">2326</span> return true;<a name="line.2326"></a> -<span class="sourceLineNo">2327</span> }<a name="line.2327"></a> -<span class="sourceLineNo">2328</span> if (this.flushCheckInterval <= 0) {<a name="line.2328"></a> -<span class="sourceLineNo">2329</span> return false;<a name="line.2329"></a> -<span class="sourceLineNo">2330</span> }<a name="line.2330"></a> -<span class="sourceLineNo">2331</span> long now = EnvironmentEdgeManager.currentTime();<a name="line.2331"></a> -<span class="sourceLineNo">2332</span> if (store.timeOfOldestEdit() < now - this.flushCheckInterval) {<a name="line.2332"></a> -<span class="sourceLineNo">2333</span> if (LOG.isDebugEnabled()) {<a name="line.2333"></a> -<span class="sourceLineNo">2334</span> LOG.debug("Flush column family: " + store.getColumnFamilyName() + " of " +<a name="line.2334"></a> -<span class="sourceLineNo">2335</span> getRegionInfo().getEncodedName() + " because time of oldest edit=" +<a name="line.2335"></a> -<span class="sourceLineNo">2336</span> store.timeOfOldestEdit() + " is > " + this.flushCheckInterval + " from now =" + now);<a name="line.2336"></a> -<span class="sourceLineNo">2337</span> }<a name="line.2337"></a> -<span class="sourceLineNo">2338</span> return true;<a name="line.2338"></a> -<span class="sourceLineNo">2339</span> }<a name="line.2339"></a> -<span class="sourceLineNo">2340</span> return false;<a name="line.2340"></a> -<span class="sourceLineNo">2341</span> }<a name="line.2341"></a> -<span class="sourceLineNo">2342</span><a name="line.2342"></a> -<span class="sourceLineNo">2343</span> /**<a name="line.2343"></a> -<span class="sourceLineNo">2344</span> * Should the memstore be flushed now<a name="line.2344"></a> -<span class="sourceLineNo">2345</span> */<a name="line.2345"></a> -<span class="sourceLineNo">2346</span> boolean shouldFlush(final StringBuilder whyFlush) {<a name="line.2346"></a> -<span class="sourceLineNo">2347</span> whyFlush.setLength(0);<a name="line.2347"></a> -<span class="sourceLineNo">2348</span> // This is a rough measure.<a name="line.2348"></a> -<span class="sourceLineNo">2349</span> if (this.maxFlushedSeqId > 0<a name="line.2349"></a> -<span class="sourceLineNo">2350</span> && (this.maxFlushedSeqId + this.flushPerChanges < this.mvcc.getReadPoint())) {<a name="line.2350"></a> -<span class="sourceLineNo">2351</span> whyFlush.append("more than max edits, " + this.flushPerChanges + ", since last flush");<a name="line.2351"></a> -<span class="sourceLineNo">2352</span> return true;<a name="line.2352"></a> -<span class="sourceLineNo">2353</span> }<a name="line.2353"></a> -<span class="sourceLineNo">2354</span> long modifiedFlushCheckInterval = flushCheckInterval;<a name="line.2354"></a> -<span class="sourceLineNo">2355</span> if (getRegionInfo().getTable().isSystemTable() &&<a name="line.2355"></a> -<span class="sourceLineNo">2356</span> getRegionInfo().getReplicaId() == RegionInfo.DEFAULT_REPLICA_ID) {<a name="line.2356"></a> -<span class="sourceLineNo">2357</span> modifiedFlushCheckInterval = SYSTEM_CACHE_FLUSH_INTERVAL;<a name="line.2357"></a> -<span class="sourceLineNo">2358</span> }<a name="line.2358"></a> -<span class="sourceLineNo">2359</span> if (modifiedFlushCheckInterval <= 0) { //disabled<a name="line.2359"></a> -<span class="sourceLineNo">2360</span> return false;<a name="line.2360"></a> -<span class="sourceLineNo">2361</span> }<a name="line.2361"></a> -<span class="sourceLineNo">2362</span> long now = EnvironmentEdgeManager.currentTime();<a name="line.2362"></a> -<span class="sourceLineNo">2363</span> //if we flushed in the recent past, we don't need to do again now<a name="line.2363"></a> -<span class="sourceLineNo">2364</span> if ((now - getEarliestFlushTimeForAllStores() < modifiedFlushCheckInterval)) {<a name="line.2364"></a> -<span class="sourceLineNo">2365</span> return false;<a name="line.2365"></a> -<span class="sourceLineNo">2366</span> }<a name="line.2366"></a> -<span class="sourceLineNo">2367</span> //since we didn't flush in the recent past, flush now if certain conditions<a name="line.2367"></a> -<span class="sourceLineNo">2368</span> //are met. Return true on first such memstore hit.<a name="line.2368"></a> -<span class="sourceLineNo">2369</span> for (HStore s : stores.values()) {<a name="line.2369"></a> -<span class="sourceLineNo">2370</span> if (s.timeOfOldestEdit() < now - modifiedFlushCheckInterval) {<a name="line.2370"></a> -<span class="sourceLineNo">2371</span> // we have an old enough edit in the memstore, flush<a name="line.2371"></a> -<span class="sourceLineNo">2372</span> whyFlush.append(s.toString() + " has an old edit so flush to free WALs");<a name="line.2372"></a> -<span class="sourceLineNo">2373</span> return true;<a name="line.2373"></a> -<span class="sourceLineNo">2374</span> }<a name="line.2374"></a> -<span class="sourceLineNo">2375</span> }<a name="line.2375"></a> -<span class="sourceLineNo">2376</span> return false;<a name="line.2376"></a> -<span class="sourceLineNo">2377</span> }<a name="line.2377"></a> -<span class="sourceLineNo">2378</span><a name="line.2378"></a> -<span class="sourceLineNo">2379</span> /**<a name="line.2379"></a> -<span class="sourceLineNo">2380</span> * Flushing all stores.<a name="line.2380"></a> -<span class="sourceLineNo">2381</span> * @see #internalFlushcache(Collection, MonitoredTask, boolean, FlushLifeCycleTracker)<a name="line.2381"></a> -<span class="sourceLineNo">2382</span> */<a name="line.2382"></a> -<span class="sourceLineNo">2383</span> private FlushResult internalFlushcache(MonitoredTask status) throws IOException {<a name="line.2383"></a> -<span class="sourceLineNo">2384</span> return internalFlushcache(stores.values(), status, false, FlushLifeCycleTracker.DUMMY);<a name="line.2384"></a> -<span class="sourceLineNo">2385</span> }<a name="line.2385"></a> -<span class="sourceLineNo">2386</span><a name="line.2386"></a> -<span class="sourceLineNo">2387</span> /**<a name="line.2387"></a> -<span class="sourceLineNo">2388</span> * Flushing given stores.<a name="line.2388"></a> -<span class="sourceLineNo">2389</span> * @see #internalFlushcache(WAL, long, Collection, MonitoredTask, boolean, FlushLifeCycleTracker)<a name="line.2389"></a> -<span class="sourceLineNo">2390</span> */<a name="line.2390"></a> -<span class="sourceLineNo">2391</span> private FlushResultImpl internalFlushcache(Collection<HStore> storesToFlush, MonitoredTask status,<a name="line.2391"></a> -<span class="sourceLineNo">2392</span> boolean writeFlushWalMarker, FlushLifeCycleTracker tracker) throws IOException {<a name="line.2392"></a> -<span class="sourceLineNo">2393</span> return internalFlushcache(this.wal, HConstants.NO_SEQNUM, storesToFlush, status,<a name="line.2393"></a> -<span class="sourceLineNo">2394</span> writeFlushWalMarker, tracker);<a name="line.2394"></a> -<span class="sourceLineNo">2395</span> }<a name="line.2395"></a> -<span class="sourceLineNo">2396</span><a name="line.2396"></a> -<span class="sourceLineNo">2397</span> /**<a name="line.2397"></a> -<span class="sourceLineNo">2398</span> * Flush the memstore. Flushing the memstore is a little tricky. We have a lot of updates in the<a name="line.2398"></a> -<span class="sourceLineNo">2399</span> * memstore, all of which have also been written to the wal. We need to write those updates in the<a name="line.2399"></a> -<span class="sourceLineNo">2400</span> * memstore out to disk, while being able to process reads/writes as much as possible during the<a name="line.2400"></a> -<span class="sourceLineNo">2401</span> * flush operation.<a name="line.2401"></a> -<span class="sourceLineNo">2402</span> * <p><a name="line.2402"></a> -<span class="sourceLineNo">2403</span> * This method may block for some time. Every time you call it, we up the regions sequence id even<a name="line.2403"></a> -<span class="sourceLineNo">2404</span> * if we don't flush; i.e. the returned region id will be at least one larger than the last edit<a name="line.2404"></a> -<span class="sourceLineNo">2405</span> * applied to this region. The returned id does not refer to an actual edit. The returned id can<a name="line.2405"></a> -<span class="sourceLineNo">2406</span> * be used for say installing a bulk loaded file just ahead of the last hfile that was the result<a name="line.2406"></a> -<span class="sourceLineNo">2407</span> * of this flush, etc.<a name="line.2407"></a> -<span class="sourceLineNo">2408</span> * @param wal Null if we're NOT to go via wal.<a name="line.2408"></a> -<span class="sourceLineNo">2409</span> * @param myseqid The seqid to use if <code>wal</code> is null writing out flush file.<a name="line.2409"></a> -<span class="sourceLineNo">2410</span> * @param storesToFlush The list of stores to flush.<a name="line.2410"></a> -<span class="sourceLineNo">2411</span> * @return object describing the flush's state<a name="line.2411"></a> -<span class="sourceLineNo">2412</span> * @throws IOException general io exceptions<a name="line.2412"></a> -<span class="sourceLineNo">2413</span> * @throws DroppedSnapshotException Thrown when replay of WAL is required.<a name="line.2413"></a> -<span class="sourceLineNo">2414</span> */<a name="line.2414"></a> -<span class="sourceLineNo">2415</span> protected FlushResultImpl internalFlushcache(WAL wal, long myseqid,<a name="line.2415"></a> -<span class="sourceLineNo">2416</span> Collection<HStore> storesToFlush, MonitoredTask status, boolean writeFlushWalMarker,<a name="line.2416"></a> -<span class="sourceLineNo">2417</span> FlushLifeCycleTracker tracker) throws IOException {<a name="line.2417"></a> -<span class="sourceLineNo">2418</span> PrepareFlushResult result =<a name="line.2418"></a> -<span class="sourceLineNo">2419</span> internalPrepareFlushCache(wal, myseqid, storesToFlush, status, writeFlushWalMarker, tracker);<a name="line.2419"></a> -<span class="sourceLineNo">2420</span> if (result.result == null) {<a name="line.2420"></a> -<span class="sourceLineNo">2421</span> return internalFlushCacheAndCommit(wal, status, result, storesToFlush);<a name="line.2421"></a> -<span class="sourceLineNo">2422</span> } else {<a name="line.2422"></a> -<span class="sourceLineNo">2423</span> return result.result; // early exit due to failure from prepare stage<a name="line.2423"></a> -<span class="sourceLineNo">2424</span> }<a name="line.2424"></a> -<span class="sourceLineNo">2425</span> }<a name="line.2425"></a> -<span class="sourceLineNo">2426</span><a name="line.2426"></a> -<span class="sourceLineNo">2427</span> @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="DLS_DEAD_LOCAL_STORE",<a name="line.2427"></a> -<span class="sourceLineNo">2428</span> justification="FindBugs seems confused about trxId")<a name="line.2428"></a> -<span class="sourceLineNo">2429</span> protected PrepareFlushResult internalPrepareFlushCache(WAL wal, long myseqid,<a name="line.2429"></a> -<span class="sourceLineNo">2430</span> Collection<HStore> storesToFlush, MonitoredTask status, boolean writeFlushWalMarker,<a name="line.2430"></a> -<span class="sourceLineNo">2431</span> FlushLifeCycleTracker tracker) throws IOException {<a name="line.2431"></a> -<span class="sourceLineNo">2432</span> if (this.rsServices != null && this.rsServices.isAborted()) {<a name="line.2432"></a> -<span class="sourceLineNo">2433</span> // Don't flush when server aborting, it's unsafe<a name="line.2433"></a> -<span class="sourceLineNo">2434</span> throw new IOException("Aborting flush because server is aborted...");<a name="line.2434"></a> -<span class="sourceLineNo">2435</span> }<a name="line.2435"></a> -<span class="sourceLineNo">2436</span> final long startTime = EnvironmentEdgeManager.currentTime();<a name="line.2436"></a> -<span class="sourceLineNo">2437</span> // If nothing to flush, return, but return with a valid unused sequenceId.<a name="line.2437"></a> -<span class="sourceLineNo">2438</span> // Its needed by bulk upload IIRC. It flushes until no edits in memory so it can insert a<a name="line.2438"></a> -<span class="sourceLineNo">2439</span> // bulk loaded file between memory and existing hfiles. It wants a good seqeunceId that belongs<a name="line.2439"></a> -<span class="sourceLineNo">2440</span> // to no other that it can use to associate with the bulk load. Hence this little dance below<a name="line.2440"></a> -<span class="sourceLineNo">2441</span> // to go get one.<a name="line.2441"></a> -<span class="sourceLineNo">2442</span> if (this.memStoreSizing.getDataSize() <= 0) {<a name="line.2442"></a> -<span class="sourceLineNo">2443</span> // Take an update lock so no edits can come into memory just yet.<a name="line.2443"></a> -<span class="sourceLineNo">2444</span> this.updatesLock.writeLock().lock();<a name="line.2444"></a> -<span class="sourceLineNo">2445</span> WriteEntry writeEntry = null;<a name="line.2445"></a> -<span class="sourceLineNo">2446</span> try {<a name="line.2446"></a> -<span class="sourceLineNo">2447</span> if (this.memStoreSizing.getDataSize() <= 0) {<a name="line.2447"></a> -<span class="sourceLineNo">2448</span> // Presume that if there are still no edits in the memstore, then there are no edits for<a name="line.2448"></a> -<span class="sourceLineNo">2449</span> // this region out in the WAL subsystem so no need to do any trickery clearing out<a name="line.2449"></a> -<span class="sourceLineNo">2450</span> // edits in the WAL sub-system. Up the sequence number so the resulting flush id is for<a name="line.2450"></a> -<span class="sourceLineNo">2451</span> // sure just beyond the last appended region edit and not associated with any edit<a name="line.2451"></a> -<span class="sourceLineNo">2452</span> // (useful as marker when bulk loading, etc.).<a name="line.2452"></a> -<span class="sourceLineNo">2453</span> if (wal != null) {<a name="line.2453"></a> -<span class="sourceLineNo">2454</span> writeEntry = mvcc.begin();<a name="line.2454"></a> -<span class="sourceLineNo">2455</span> long flushOpSeqId = writeEntry.getWriteNumber();<a name="line.2455"></a> -<span class="sourceLineNo">2456</span> FlushResultImpl flushResult =<a name="line.2456"></a> -<span class="sourceLineNo">2457</span> new FlushResultImpl(FlushResult.Result.CANNOT_FLUSH_MEMSTORE_EMPTY, flushOpSeqId,<a name="line.2457"></a> -<span class="sourceLineNo">2458</span> "Nothing to flush", writeFlushRequestMarkerToWAL(wal, writeFlushWalMarker));<a name="line.2458"></a> -<span class="sourceLineNo">2459</span> mvcc.completeAndWait(writeEntry);<a name="line.2459"></a> -<span class="sourceLineNo">2460</span> // Set to null so we don't complete it again down in finally block.<a name="line.2460"></a> -<span class="sourceLineNo">2461</span> writeEntry = null;<a name="line.2461"></a> -<span class="sourceLineNo">2462</span> return new PrepareFlushResult(flushResult, myseqid);<a name="line.2462"></a> -<span class="sourceLineNo">2463</span> } else {<a name="line.2463"></a> -<span class="sourceLineNo">2464</span> return new PrepareFlushResult(new FlushResultImpl(<a name="line.2464"></a> -<span class="sourceLineNo">2465</span> FlushResult.Result.CANNOT_FLUSH_MEMSTORE_EMPTY, "Nothing to flush", false), myseqid);<a name="line.2465"></a> -<span class="sourceLineNo">2466</span> }<a name="line.2466"></a> -<span class="sourceLineNo">2467</span> }<a name="line.2467"></a> -<span class="sourceLineNo">2468</span> } finally {<a name="line.2468"></a> -<span class="sourceLineNo">2469</span> if (writeEntry != null) {<a name="line.2469"></a> -<span class="sourceLineNo">2470</span> // If writeEntry is non-null, this operation failed; the mvcc transaction failed...<a name="line.2470"></a> -<span class="sourceLineNo">2471</span> // but complete it anyways so it doesn't block the mvcc queue.<a name="line.2471"></a> -<span class="sourceLineNo">2472</span> mvcc.complete(writeEntry);<a name="line.2472"></a> +<span class="sourceLineNo">2291</span> if (coprocessorHost != null) {<a name="line.2291"></a> +<span class="sourceLineNo">2292</span> status.setStatus("Running post-flush coprocessor hooks");<a name="line.2292"></a> +<span class="sourceLineNo">2293</span> coprocessorHost.postFlush(tracker);<a name="line.2293"></a> +<span class="sourceLineNo">2294</span> }<a name="line.2294"></a> +<span class="sourceLineNo">2295</span><a name="line.2295"></a> +<span class="sourceLineNo">2296</span> if(fs.isFlushSucceeded()) {<a name="line.2296"></a> +<span class="sourceLineNo">2297</span> flushesQueued.reset();<a name="line.2297"></a> +<span class="sourceLineNo">2298</span> }<a name="line.2298"></a> +<span class="sourceLineNo">2299</span><a name="line.2299"></a> +<span class="sourceLineNo">2300</span> status.markComplete("Flush successful");<a name="line.2300"></a> +<span class="sourceLineNo">2301</span> return fs;<a name="line.2301"></a> +<span class="sourceLineNo">2302</span> } finally {<a name="line.2302"></a> +<span class="sourceLineNo">2303</span> synchronized (writestate) {<a name="line.2303"></a> +<span class="sourceLineNo">2304</span> writestate.flushing = false;<a name="line.2304"></a> +<span class="sourceLineNo">2305</span> this.writestate.flushRequested = false;<a name="line.2305"></a> +<span class="sourceLineNo">2306</span> writestate.notifyAll();<a name="line.2306"></a> +<span class="sourceLineNo">2307</span> }<a name="line.2307"></a> +<span class="sourceLineNo">2308</span> }<a name="line.2308"></a> +<span class="sourceLineNo">2309</span> } finally {<a name="line.2309"></a> +<span class="sourceLineNo">2310</span> lock.readLock().unlock();<a name="line.2310"></a> +<span class="sourceLineNo">2311</span> LOG.debug("Flush status journal:\n\t" + status.prettyPrintJournal());<a name="line.2311"></a> +<span class="sourceLineNo">2312</span> status.cleanup();<a name="line.2312"></a> +<span class="sourceLineNo">2313</span> }<a name="line.2313"></a> +<span class="sourceLineNo">2314</span> }<a name="line.2314"></a> +<span class="sourceLineNo">2315</span><a name="line.2315"></a> +<span class="sourceLineNo">2316</span> /**<a name="line.2316"></a> +<span class="sourceLineNo">2317</span> * Should the store be flushed because it is old enough.<a name="line.2317"></a> +<span class="sourceLineNo">2318</span> * <p><a name="line.2318"></a> +<span class="sourceLineNo">2319</span> * Every FlushPolicy should call this to determine whether a store is old enough to flush (except<a name="line.2319"></a> +<span class="sourceLineNo">2320</span> * that you always flush all stores). Otherwise the method will always<a name="line.2320"></a> +<span class="sourceLineNo">2321</span> * returns true which will make a lot of flush requests.<a name="line.2321"></a> +<span class="sourceLineNo">2322</span> */<a name="line.2322"></a> +<span class="sourceLineNo">2323</span> boolean shouldFlushStore(HStore store) {<a name="line.2323"></a> +<span class="sourceLineNo">2324</span> long earliest = this.wal.getEarliestMemStoreSeqNum(getRegionInfo().getEncodedNameAsBytes(),<a name="line.2324"></a> +<span class="sourceLineNo">2325</span> store.getColumnFamilyDescriptor().getName()) - 1;<a name="line.2325"></a> +<span class="sourceLineNo">2326</span> if (earliest > 0 && earliest + flushPerChanges < mvcc.getReadPoint()) {<a name="line.2326"></a> +<span class="sourceLineNo">2327</span> if (LOG.isDebugEnabled()) {<a name="line.2327"></a> +<span class="sourceLineNo">2328</span> LOG.debug("Flush column family " + store.getColumnFamilyName() + " of " +<a name="line.2328"></a> +<span class="sourceLineNo">2329</span> getRegionInfo().getEncodedName() + " because unflushed sequenceid=" + earliest +<a name="line.2329"></a> +<span class="sourceLineNo">2330</span> " is > " + this.flushPerChanges + " from current=" + mvcc.getReadPoint());<a name="line.2330"></a> +<span class="sourceLineNo">2331</span> }<a name="line.2331"></a> +<span class="sourceLineNo">2332</span> return true;<a name="line.2332"></a> +<span class="sourceLineNo">2333</span> }<a name="line.2333"></a> +<span class="sourceLineNo">2334</span> if (this.flushCheckInterval <= 0) {<a name="line.2334"></a> +<span class="sourceLineNo">2335</span> return false;<a name="line.2335"></a> +<span class="sourceLineNo">2336</span> }<a name="line.2336"></a> +<span class="sourceLineNo">2337</span> long now = EnvironmentEdgeManager.currentTime();<a name="line.2337"></a> +<span class="sourceLineNo">2338</span> if (store.timeOfOldestEdit() < now - this.flushCheckInterval) {<a name="line.2338"></a> +<span class="sourceLineNo">2339</span> if (LOG.isDebugEnabled()) {<a name="line.2339"></a> +<span class="sourceLineNo">2340</span> LOG.debug("Flush column family: " + store.getColumnFamilyName() + " of " +<a name="line.2340"></a> +<span class="sourceLineNo">2341</span> getRegionInfo().getEncodedName() + " because time of oldest edit=" +<a name="line.2341"></a> +<span class="sourceLineNo">2342</span> store.timeOfOldestEdit() + " is > " + this.flushCheckInterval + " from now =" + now);<a name="line.2342"></a> +<span class="sourceLineNo">2343</span> }<a name="line.2343"></a> +<span class="sourceLineNo">2344</span> return true;<a name="line.2344"></a> +<span class="sourceLineNo">2345</span> }<a name="line.2345"></a> +<span class="sourceLineNo">2346</span> return false;<a name="line.2346"></a> +<span class="sourceLineNo">2347</span> }<a name="line.2347"></a> +<span class="sourceLineNo">2348</span><a name="line.2348"></a> +<span class="sourceLineNo">2349</span> /**<a name="line.2349"></a> +<span class="sourceLineNo">2350</span> * Should the memstore be flushed now<a name="line.2350"></a> +<span class="sourceLineNo">2351</span> */<a name="line.2351"></a> +<span class="sourceLineNo">2352</span> boolean shouldFlush(final StringBuilder whyFlush) {<a name="line.2352"></a> +<span class="sourceLineNo">2353</span> whyFlush.setLength(0);<a name="line.2353"></a> +<span class="sourceLineNo">2354</span> // This is a rough measure.<a name="line.2354"></a> +<span class="sourceLineNo">2355</span> if (this.maxFlushedSeqId > 0<a name="line.2355"></a> +<span class="sourceLineNo">2356</span> && (this.maxFlushedSeqId + this.flushPerChanges < this.mvcc.getReadPoint())) {<a name="line.2356"></a> +<span class="sourceLineNo">2357</span> whyFlush.append("more than max edits, " + this.flushPerChanges + ", since last flush");<a name="line.2357"></a> +<span class="sourceLineNo">2358</span> return true;<a name="line.2358"></a> +<span class="sourceLineNo">2359</span> }<a name="line.2359"></a> +<span class="sourceLineNo">2360</span> long modifiedFlushCheckInterval = flushCheckInterval;<a name="line.2360"></a> +<span class="sourceLineNo">2361</span> if (getRegionInfo().getTable().isSystemTable() &&<a name="line.2361"></a> +<span class="sourceLineNo">2362</span> getRegionInfo().getReplicaId() == RegionInfo.DEFAULT_REPLICA_ID) {<a name="line.2362"></a> +<span class="sourceLineNo">2363</span> modifiedFlushCheckInterval = SYSTEM_CACHE_FLUSH_INTERVAL;<a name="line.2363"></a> +<span class="sourceLineNo">2364</span> }<a name="line.2364"></a> +<span class="sourceLineNo">2365</span> if (modifiedFlushCheckInterval <= 0) { //disabled<a name="line.2365"></a> +<span class="sourceLineNo">2366</span> return false;<a name="line.2366"></a> +<span class="sourceLineNo">2367</span> }<a name="line.2367"></a> +<span class="sourceLineNo">2368</span> long now = EnvironmentEdgeManager.currentTime();<a name="line.2368"></a> +<span class="sourceLineNo">2369</span> //if we flushed in the recent past, we don't need to do again now<a name="line.2369"></a> +<span class="sourceLineNo">2370</span> if ((now - getEarliestFlushTimeForAllStores() < modifiedFlushCheckInterval)) {<a name="line.2370"></a> +<span class="sourceLineNo">2371</span> return false;<a name="line.2371"></a> +<span class="sourceLineNo">2372</span> }<a name="line.2372"></a> +<span class="sourceLineNo">2373</span> //since we didn't flush in the recent past, flush now if certain conditions<a name="line.2373"></a> +<span class="sourceLineNo">2374</span> //are met. Return true on first such memstore hit.<a name="line.2374"></a> +<span class="sourceLineNo">2375</span> for (HStore s : stores.values()) {<a name="line.2375"></a> +<span class="sourceLineNo">2376</span> if (s.timeOfOldestEdit() < now - modifiedFlushCheckInterval) {<a name="line.2376"></a> +<span class="sourceLineNo">2377</span> // we have an old enough edit in the memstore, flush<a name="line.2377"></a> +<span class="sourceLineNo">2378</span> whyFlush.append(s.toString() + " has an old edit so flush to free WALs");<a name="line.2378"></a> +<span class="sourceLineNo">2379</span> return true;<a name="line.2379"></a> +<span class="sourceLineNo">2380</span> }<a name="line.2380"></a> +<span class="sourceLineNo">2381</span> }<a name="line.2381"></a> +<span class="sourceLineNo">2382</span> return false;<a name="line.2382"></a> +<span class="sourceLineNo">2383</span> }<a name="line.2383"></a> +<span class="sourceLineNo">2384</span><a name="line.2384"></a> +<span class="sourceLineNo">2385</span> /**<a name="line.2385"></a> +<span class="sourceLineNo">2386</span> * Flushing all stores.<a name="line.2386"></a> +<span class="sourceLineNo">2387</span> * @see #internalFlushcache(Collection, MonitoredTask, boolean, FlushLifeCycleTracker)<a name="line.2387"></a> +<span class="sourceLineNo">2388</span> */<a name="line.2388"></a> +<span class="sourceLineNo">2389</span> private FlushResult internalFlushcache(MonitoredTask status) throws IOException {<a name="line.2389"></a> +<span class="sourceLineNo">2390</span> return internalFlushcache(stores.values(), status, false, FlushLifeCycleTracker.DUMMY);<a name="line.2390"></a> +<span class="sourceLineNo">2391</span> }<a name="line.2391"></a> +<span class="sourceLineNo">2392</span><a name="line.2392"></a> +<span class="sourceLineNo">2393</span> /**<a name="line.2393"></a> +<span class="sourceLineNo">2394</span> * Flushing given stores.<a name="line.2394"></a> +<span class="sourceLineNo">2395</span> * @see #internalFlushcache(WAL, long, Collection, MonitoredTask, boolean, FlushLifeCycleTracker)<a name="line.2395"></a> +<span class="sourceLineNo">2396</span> */<a name="line.2396"></a> +<span class="sourceLineNo">2397</span> private FlushResultImpl internalFlushcache(Collection<HStore> storesToFlush, MonitoredTask status,<a name="line.2397"></a> +<span class="sourceLineNo">2398</span> boolean writeFlushWalMarker, FlushLifeCycleTracker tracker) throws IOException {<a name="line.2398"></a> +<span class="sourceLineNo">2399</span> return internalFlushcache(this.wal, HConstants.NO_SEQNUM, storesToFlush, status,<a name="line.2399"></a> +<span class="sourceLineNo">2400</span> writeFlushWalMarker, tracker);<a name="line.2400"></a> +<span class="sourceLineNo">2401</span> }<a name="line.2401"></a> +<span class="sourceLineNo">2402</span><a name="line.2402"></a> +<span class="sourceLineNo">2403</span> /**<a name="line.2403"></a> +<span class="sourceLineNo">2404</span> * Flush the memstore. Flushing the memstore is a little tricky. We have a lot of updates in the<a name="line.2404"></a> +<span class="sourceLineNo">2405</span> * memstore, all of which have also been written to the wal. We need to write those updates in the<a name="line.2405"></a> +<span class="sourceLineNo">2406</span> * memstore out to disk, while being able to process reads/writes as much as possible during the<a name="line.2406"></a> +<span class="sourceLineNo">2407</span> * flush operation.<a name="line.2407"></a> +<span class="sourceLineNo">2408</span> * <p><a name="line.2408"></a> +<span class="sourceLineNo">2409</span> * This method may block for some time. Every time you call it, we up the regions sequence id even<a name="line.2409"></a> +<span class="sourceLineNo">2410</span> * if we don't flush; i.e. the returned region id will be at least one larger than the last edit<a name="line.2410"></a> +<span class="sourceLineNo">2411</span> * applied to this region. The returned id does not refer to an actual edit. The returned id can<a name="line.2411"></a> +<span class="sourceLineNo">2412</span> * be used for say installing a bulk loaded file just ahead of the last hfile that was the result<a name="line.2412"></a> +<span class="sourceLineNo">2413</span> * of this flush, etc.<a name="line.2413"></a> +<span class="sourceLineNo">2414</span> * @param wal Null if we're NOT to go via wal.<a name="line.2414"></a> +<span class="sourceLineNo">2415</span> * @param myseqid The seqid to use if <code>wal</code> is null writing out flush file.<a name="line.2415"></a> +<span class="sourceLineNo">2416</span> * @param storesToFlush The list of stores to flush.<a name="line.2416"></a> +<span class="sourceLineNo">2417</span> * @return object describing the flush's state<a name="line.2417"></a> +<span class="sourceLineNo">2418</span> * @throws IOException general io exceptions<a name="line.2418"></a> +<span class="sourceLineNo">2419</span> * @throws DroppedSnapshotException Thrown when replay of WAL is required.<a name="line.2419"></a> +<span class="sourceLineNo">2420</span> */<a name="line.2420"></a> +<span class="sourceLineNo">2421</span> protected FlushResultImpl internalFlushcache(WAL wal, long myseqid,<a name="line.2421"></a> +<span class="sourceLineNo">2422</span> Collection<HStore> storesToFlush, MonitoredTask status, boolean writeFlushWalMarker,<a name="line.2422"></a> +<span class="sourceLineNo">2423</span> FlushLifeCycleTracker tracker) throws IOException {<a name="line.2423"></a> +<span class="sourceLineNo">2424</span> PrepareFlushResult result =<a name="line.2424"></a> +<span class="sourceLineNo">2425</span> internalPrepareFlushCache(wal, myseqid, storesToFlush, status, writeFlushWalMarker, tracker);<a name="line.2425"></a> +<span class="sourceLineNo">2426</span> if (result.result == null) {<a name="line.2426"></a> +<span class="sourceLineNo">2427</span> return internalFlushCacheAndCommit(wal, status, result, storesToFlush);<a name="line.2427"></a> +<span class="sourceLineNo">2428</span> } else {<a name="line.2428"></a> +<span class="sourceLineNo">2429</span> return result.result; // early exit due to failure from prepare stage<a name="line.2429"></a> +<span class="sourceLineNo">2430</span> }<a name="line.2430"></a> +<span class="sourceLineNo">2431</span> }<a name="line.2431"></a> +<span class="sourceLineNo">2432</span><a name="line.2432"></a> +<span class="sourceLineNo">2433</span> @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="DLS_DEAD_LOCAL_STORE",<a name="line.2433"></a> +<span class="sourceLineNo">2434</span> justification="FindBugs seems confused about trxId")<a name="line.2434"></a> +<span class="sourceLineNo">2435</span> protected PrepareFlushResult internalPrepareFlushCache(WAL wal, long myseqid,<a name="line.2435"></a> +<span class="sourceLineNo">2436</span> Collection<HStore> storesToFlush, MonitoredTask status, boolean writeFlushWalMarker,<a name="line.2436"></a> +<span class="sourceLineNo">2437</span> FlushLifeCycleTracker tracker) throws IOException {<a name="line.2437"></a> +<span class="sourceLineNo">2438</span> if (this.rsServices != null && this.rsServices.isAborted()) {<a name="line.2438"></a> +<span class="sourceLineNo">2439</span> // Don't flush when server aborting, it's unsafe<a name="line.2439"></a> +<span class="sourceLineNo">2440</span> throw new IOException("Aborting flush because server is aborted...");<a name="line.2440"></a> +<span class="sourceLineNo">2441</span> }<a name="line.2441"></a> +<span class="sourceLineNo">2442</span> final long startTime = EnvironmentEdgeManager.currentTime();<a name="line.2442"></a> +<span class="sourceLineNo">2443</span> // If nothing to flush, return, but return with a valid unused sequenceId.<a name="line.2443"></a> +<span class="sourceLineNo">2444</span> // Its needed by bulk upload IIRC. It flushes until no edits in memory so it can insert a<a name="line.2444"></a> +<span class="sourceLineNo">2445</span> // bulk loaded file between memory and existing hfiles. It wants a good seqeunceId that belongs<a name="line.2445"></a> +<span class="sourceLineNo">2446</span> // to no other that it can use to associate with the bulk load. Hence this little dance below<a name="line.2446"></a> +<span class="sourceLineNo">2447</span> // to go get one.<a name="line.2447"></a> +<span class="sourceLineNo">2448</span> if (this.memStoreSizing.getDataSize() <= 0) {<a name="line.2448"></a> +<span class="sourceLineNo">2449</span> // Take an update lock so no edits can come into memory just yet.<a name="line.2449"></a> +<span class="sourceLineNo">2450</span> this.updatesLock.writeLock().lock();<a name="line.2450"></a> +<span class="sourceLineNo">2451</span> WriteEntry writeEntry = null;<a name="line.2451"></a> +<span class="sourceLineNo">2452</span> try {<a name="line.2452"></a> +<span class="sourceLineNo">2453</span> if (this.memStoreSizing.getDataSize() <= 0) {<a name="line.2453"></a> +<span class="sourceLineNo">2454</span> // Presume that if there are still no edits in the memstore, then there are no edits for<a name="line.2454"></a> +<span class="sourceLineNo">2455</span> // this region out in the WAL subsystem so no need to do any trickery clearing out<a name="line.2455"></a> +<span class="sourceLineNo">2456</span> // edits in the WAL sub-system. Up the sequence number so the resulting flush id is for<a name="line.2456"></a> +<span class="sourceLineNo">2457</span> // sure just beyond the last appended region edit and not associated with any edit<a name="line.2457"></a> +<span class="sourceLineNo">2458</span> // (useful as marker when bulk loading, etc.).<a name="line.2458"></a> +<span class="sourceLineNo">2459</span> if (wal != null) {<a name="line.2459"></a> +<span class="sourceLineNo">2460</span> writeEntry = mvcc.begin();<a name="line.2460"></a> +<span class="sourceLineNo">2461</span> long flushOpSeqId = writeEntry.getWriteNumber();<a name="line.2461"></a> +<span class="sourceLineNo">2462</span> FlushResultImpl flushResult =<a name="line.2462"></a> +<span class="sourceLineNo">2463</span> new FlushResultImpl(FlushResult.Result.CANNOT_FLUSH_MEMSTORE_EMPTY, flushOpSeqId,<a name="line.2463"></a> +<span class="sourceLineNo">2464</span> "Nothing to flush", writeFlushRequestMarkerToWAL(wal, writeFlushWalMarker));<a name="line.2464"></a> +<span class="sourceLineNo">2465</span> mvcc.completeAndWait(writeEntry);<a name="line.2465"></a> +<span class="sourceLineNo">2466</span> // Set to null so we don't complete it again down in finally block.<a name="line.2466"></a> +<span class="sourceLineNo">2467</span> writeEntry = null;<a name="line.2467"></a> +<span class="sourceLineNo">2468</span> return new PrepareFlushResult(flushResult, myseqid);<a name="line.2468"></a> +<span class="sourceLineNo">2469</span> } else {<a name="line.2469"></a> +<span class="sourceLineNo">2470</span> return new PrepareFlushResult(new FlushResultImpl(<a name="line.2470"></a> +<span class="sourceLineNo">2471</span> FlushResult.Result.CANNOT_FLUSH_MEMSTORE_EMPTY, "Nothing to flush", false), myseqid);<a name="line.2471"></a> +<span class="sourceLineNo">2472</span> }<a name="line.2472"></a> <span class="sourceLineNo">2473</span> }<a name="line.2473"></a> -<span class="sourceLineNo">2474</span> this.updatesLock.writeLock().unlock();<a name="line.2474"></a> -<span class="sourceLineNo">2475</span> }<a name="line.2475"></a> -<span class="sourceLineNo">2476</span> }<a name="line.2476"></a> -<span class="sourceLineNo">2477</span> logFatLineOnFlush(storesToFlush, myseqid);<a name="line.2477"></a> -<span class="sourceLineNo">2478</span> // Stop updates while we snapshot the memstore of all of these regions' stores. We only have<a name="line.2478"></a> -<span class="sourceLineNo">2479</span> // to do this for a moment. It is quick. We also set the memstore size to zero here before we<a name="line.2479"></a> -<span class="sourceLineNo">2480</span> // allow updates again so its value will represent the size of the updates received<a name="line.2480"></a> -<span class="sourceLineNo">2481</span> // during flush<a name="line.2481"></a> -<span class="sourceLineNo">2482</span><a name="line.2482"></a> -<span class="sourceLineNo">2483</span> // We have to take an update lock during snapshot, or else a write could end up in both snapshot<a name="line.2483"></a> -<span class="sourceLineNo">2484</span> // and memstore (makes it difficult to do atomic rows then)<a name="line.2484"></a> -<span class="sourceLineNo">2485</span> status.setStatus("Obtaining lock to block concurrent updates");<a name="line.2485"></a> -<span class="sourceLineNo">2486</span> // block waiting for the lock for internal flush<a name="line.2486"></a> -<span class="sourceLineNo">2487</span> this.updatesLock.writeLock().lock();<a name="line.2487"></a> -<span class="sourceLineNo">2488</span> status.setStatus("Preparing flush snapshotting stores in " + getRegionInfo().getEncodedName());<a name="line.2488"></a> -<span class="sourceLineNo">2489</span> MemStoreSizing totalSizeOfFlushableStores = new NonThreadSafeMemStoreSizing();<a name="line.2489"></a> -<span class="sourceLineNo">2490</span><a name="line.2490"></a> -<span class="sourceLineNo">2491</span> Map<byte[], Long> flushedFamilyNamesToSeq = new HashMap<>();<a name="line.2491"></a> -<span class="sourceLineNo">2492</span> for (HStore store : storesToFlush) {<a name="line.2492"></a> -<span class="sourceLineNo">2493</span> flushedFamilyNamesToSeq.put(store.getColumnFamilyDescriptor().getName(),<a name="line.2493"></a> -<span class="sourceLineNo">2494</span> store.preFlushSeqIDEstimation());<a name="line.2494"></a> -<span class="sourceLineNo">2495</span> }<a name="line.2495"></a> +<span class="sourceLineNo">2474</span> } finally {<a name="line.2474"></a> +<span class="sourceLineNo">2475</span> if (writeEntry != null) {<a name="line.2475"></a> +<span class="sourceLineNo">2476</span> // If writeEntry is non-null, this operation failed; the mvcc transaction failed...<a name="line.2476"></a> +<span class="sourceLineNo">2477</span> // but complete it anyways so it doesn't block the mvcc queue.<a name="line.2477"></a> +<span class="sourceLineNo">2478</span> mvcc.complete(writeEntry);<a name="line.2478"></a> +<span class="sourceLineNo">2479</span> }<a name="line.2479"></a> +<span class="sourceLineNo">2480</span> this.updatesLock.writeLock().unlock();<a name="line.2480"></a> +<span class="sourceLineNo">2481</span> }<a name="line.2481"></a> +<span class="sourceLineNo">2482</span> }<a name="line.2482"></a> +<span class="sourceLineNo">2483</span> logFatLineOnFlush(storesToFlush, myseqid);<a name="line.2483"></a> +<span class="sourceLineNo">2484</span> // Stop updates while we snapshot the memstore of all of these regions' stores. We only have<a name="line.2484"></a> +<span class="sourceLineNo">2485</span> // to do this for a moment. It is quick. We also set the memstore size to zero here before we<a name="line.2485"></a> +<span class="sourceLineNo">2486</span> // allow updates again so its value will represent the size of the updates received<a name="line.2486"></a> +<span class="sourceLineNo">2487</span> // during flush<a name="line.2487"></a> +<span class="sourceLineNo">2488</span><a name="line.2488"></a> +<span class="sourceLineNo">2489</span> // We have to take an update lock during snapshot, or else a write could end up in both snapshot<a name="line.2489"></a> +<span class="sourceLineNo">2490</span> // and memstore (makes it difficult to do atomic rows then)<a name="line.2490"></a> +<span class="sourceLineNo">2491</span> status.setStatus("Obtaining lock to block concurrent updates");<a name="line.2491"></a> +<span class="sourceLineNo">2492</span> // block waiting for the lock for internal flush<a name="line.2492"></a> +<span class="sourceLineNo">2493</span> this.updatesLock.writeLock().lock();<a name="line.2493"></a> +<span class="sourceLineNo">2494</span> status.setStatus("Preparing flush snapshotting stores in " + getRegionInfo().getEncodedName());<a name="line.2494"></a> +<span class="sourceLineNo">2495</span> MemStoreSizing totalSizeOfFlushableStores = new NonThreadSafeMemStoreSizing();<a name="line.2495"></a> <span class="sourceLineNo">2496</span><a name="line.2496"></a> -<span class="sourceLineNo">2497</span> TreeMap<byte[], StoreFlushContext> storeFlushCtxs = new TreeMap<>(Bytes.BYTES_COMPARATOR);<a name="line.2497"></a> -<span class="sourceLineNo">2498</span> TreeMap<byte[], List<Path>> committedFiles = new TreeMap<>(Bytes.BYTES_COMPARATOR);<a name="line.2498"></a> -<span class="sourceLineNo">2499</span> TreeMap<byte[], MemStoreSize> storeFlushableSize = new TreeMap<>(Bytes.BYTES_COMPARATOR);<a name="line.2499"></a> -<span class="sourceLineNo">2500</span> // The sequence id of this flush operation which is used to log FlushMarker and pass to<a name="line.2500"></a> -<span class="sourceLineNo">2501</span> // createFlushContext to use as the store file's sequence id. It can be in advance of edits<a name="line.2501"></a> -<span class="sourceLineNo">2502</span> // still in the memstore, edits that are in other column families yet to be flushed.<a name="line.2502"></a> -<span class="sourceLineNo">2503</span> long flushOpSeqId = HConstants.NO_SEQNUM;<a name="line.2503"></a> -<span class="sourceLineNo">2504</span> // The max flushed sequence id after this flush operation completes. All edits in memstore<a name="line.2504"></a> -<span class="sourceLineNo">2505</span> // will be in advance of this sequence id.<a name="line.2505"></a> -<span class="sourceLineNo">2506</span> long flushedSeqId = HConstants.NO_SEQNUM;<a name="line.2506"></a> -<span class="sourceLineNo">2507</span> byte[] encodedRegionName = getRegionInfo().getEncodedNameAsBytes();<a name="line.2507"></a> -<span class="sourceLineNo">2508</span> try {<a name="line.2508"></a> -<span class="sourceLineNo">2509</span> if (wal != null) {<a name="line.2509"></a> -<span class="sourceLineNo">2510</span> Long earliestUnflushedSequenceIdForTheRegion =<a name="line.2510"></a> -<span class="sourceLineNo">2511</span> wal.startCacheFlush(encodedRegionName, flushedFamilyNamesToSeq);<a name="line.2511"></a> -<span class="sourceLineNo">2512</span> if (earliestUnflushedSequenceIdForTheRegion == null) {<a name="line.2512"></a> -<span class="sourceLineNo">2513</span> // This should never happen. This is how startCacheFlush signals flush cannot proceed.<a name="line.2513"></a> -<span class="sourceLineNo">2514</span> String msg = this.getRegionInfo().getEncodedName() + " flush aborted; WAL closing.";<a name="line.2514"></a> -<span class="sourceLineNo">2515</span> status.setStatus(msg);<a name="line.2515"></a> -<span class="sourceLineNo">2516</span> return new PrepareFlushResult(<a name="line.2516"></a> -<span class="sourceLineNo">2517</span> new FlushResultImpl(FlushResult.Result.CANNOT_FLUSH, msg, false),<a name="line.2517"></a> -<span class="sourceLineNo">2518</span> myseqid);<a name="line.2518"></a> -<span class="sourceLineNo">2519</span> }<a name="line.2519"></a> -<span class="sourceLineNo">2520</span> flushOpSeqId = getNextSequenceId(wal);<a name="line.2520"></a> -<span class="sourceLineNo">2521</span> // Back up 1, minus 1 from oldest sequence id in memstore to get last 'flushed' edit<a name="line.2521"></a> -<span class="sourceLineNo">2522</span> flushedSeqId =<a name="line.2522"></a> -<span class="sourceLineNo">2523</span> earliestUnflushedSequenceIdForTheRegion.longValue() == HConstants.NO_SEQNUM?<a name="line.2523"></a> -<span class="sourceLineNo">2524</span> flushOpSeqId: earliestUnflushedSequenceIdForTheRegion.longValue() - 1;<a name="line.2524"></a> -<span class="sourceLineNo">2525</span> } else {<a name="line.2525"></a> -<span class="sourceLineNo">2526</span> // use the provided sequence Id as WAL is not being used for this flush.<a name="line.2526"></a> -<span class="sourceLineNo">2527</span> flushedSeqId = flushOpSeqId = myseqid;<a name="line.2527"></a> -<span class="sourceLineNo">2528</span> }<a name="line.2528"></a> -<span class="sourceLineNo">2529</span><a name="line.2529"></a> -<span class="sourceLineNo">2530</span> for (HStore s : storesToFlush) {<a name="line.2530"></a> -<span class="sourceLineNo">2531</span> storeFlushCtxs.put(s.getColumnFamilyDescriptor().getName(),<a name="line.2531"></a> -<span class="sourceLineNo">2532</span> s.createFlushContext(flushOpSeqId, tracker));<a name="line.2532"></a> -<span class="sourceLineNo">2533</span> // for writing stores to WAL<a name="line.2533"></a> -<span class="sourceLineNo">2534</span> committedFiles.put(s.getColumnFamilyDescriptor().getName(), null);<a name="line.2534"></a> -<span class="sourceLineNo">2535</span> }<a name="line.2535"></a> -<span class="sourceLineNo">2536</span><a name="line.2536"></a> -<span class="sourceLineNo">2537</span> // write the snapshot start to WAL<a name="line.2537"></a> -<span class="sourceLineNo">2538</span> if (wal != null && !writestate.readOnly) {<a name="line.2538"></a> -<span class="sourceLineNo">2539</span> FlushDescriptor desc = ProtobufUtil.toFlushDescriptor(FlushAction.START_FLUSH,<a name="line.2539"></a> -<span class="sourceLineNo">2540</span> getRegionInfo(), flushOpSeqId, committedFiles);<a name="line.2540"></a> -<span class="sourceLineNo">2541</span> // No sync. Sync is below where no updates lock and we do FlushAction.COMMIT_FLUSH<a name="line.2541"></a> -<span class="sourceLineNo">2542</span> WALUtil.writeFlushMarker(wal, this.getReplicationScope(), getRegionInfo(), desc, false,<a name="line.2542"></a> -<span class="sourceLineNo">2543</span> mvcc);<a name="line.2543"></a> -<span class="sourceLineNo">2544</span> }<a name="line.2544"></a> -<span class="sourceLineNo">2545</span><a name="line.2545"></a> -<span class="sourceLineNo">2546</span> // Prepare flush (take a snapshot)<a name="line.2546"></a> -<span class="sourceLineNo">2547</span> storeFlushCtxs.forEach((name, flush) -> {<a name="line.2547"></a> -<span class="sourceLineNo">2548</span> MemStoreSize snapshotSize = flush.prepare();<a name="line.2548"></a> -<span class="sourceLineNo">2549</span> totalSizeOfFlushableStores.incMemStoreSize(snapshotSize);<a name="line.2549"></a> -<span class="sourceLineNo">2550</span> storeFlushableSize.put(name, snapshotSize);<a name="line.2550"></a> -<span class="sourceLineNo">2551</span> });<a name="line.2551"></a> -<span class="sourceLineNo">2552</span> } catch (IOException ex) {<a name="line.2552"></a> -<span class="sourceLineNo">2553</span> doAbortFlushToWAL(wal, flushOpSeqId, committedFiles);<a name="line.2553"></a> -<span class="sourceLineNo">2554</span> throw ex;<a name="line.2554"></a> -<span class="sourceLineNo">2555</span> } finally {<a name="line.2555"></a> -<span class="sourceLineNo">2556</span> this.updatesLock.writeLock().unlock();<a name="line.2556"></a> -<span class="sourceLineNo">2557</span> }<a name="line.2557"></a> -<span class="sourceLineNo">2558</span> String s = "Finished memstore snapshotting " + this + ", syncing WAL and waiting on mvcc, " +<a name="line.2558"></a> -<span class="sourceLineNo">2559</span>
<TRUNCATED>