hawq-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From benchristel <...@git.apache.org>
Subject [GitHub] incubator-hawq pull request #1379: HAWQ-1622. Cache PXF proxy UGI so that cl...
Date Sat, 07 Jul 2018 01:04:05 GMT
Github user benchristel commented on a diff in the pull request:

    https://github.com/apache/incubator-hawq/pull/1379#discussion_r200797134
  
    --- Diff: pxf/pxf-service/src/main/java/org/apache/hawq/pxf/service/UGICache.java ---
    @@ -0,0 +1,129 @@
    +package org.apache.hawq.pxf.service;
    +
    +/*
    + * Licensed to the Apache Software Foundation (ASF) under one
    + * or more contributor license agreements.  See the NOTICE file
    + * distributed with this work for additional information
    + * regarding copyright ownership.  The ASF licenses this file
    + * to you under the Apache License, Version 2.0 (the
    + * "License"); you may not use this file except in compliance
    + * with the License.  You may obtain a copy of the License at
    + *
    + *   http://www.apache.org/licenses/LICENSE-2.0
    + *
    + * Unless required by applicable law or agreed to in writing,
    + * software distributed under the License is distributed on an
    + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    + * KIND, either express or implied.  See the License for the
    + * specific language governing permissions and limitations
    + * under the License.
    + */
    +
    +import java.io.IOException;
    +import java.util.Map;
    +import java.util.concurrent.ConcurrentHashMap;
    +import java.util.concurrent.DelayQueue;
    +
    +import org.apache.commons.logging.Log;
    +import org.apache.commons.logging.LogFactory;
    +import org.apache.hadoop.fs.FileSystem;
    +import org.apache.hadoop.security.UserGroupInformation;
    +
    +public class UGICache {
    +
    +    private static final Log LOG = LogFactory.getLog(UGICache.class);
    +    private static Map<SegmentTransactionId, TimedProxyUGI> cache = new ConcurrentHashMap<>();
    +    //private static DelayQueue<TimedProxyUGI> delayQueue = new DelayQueue<>();
    +    private static DelayQueue<TimedProxyUGI>[] delayQueues = new DelayQueue<>[64];
    +    public static long UGI_CACHE_EXPIRY = 15 * 1 * 1000L; // 15 Minutes
    +
    +    public UGICache() {
    +        for (int i = 0; i < delayQueues.length; i++) {
    +            delayQueues[i] = new DelayQueue<>();
    +        }
    +    }
    +
    +    public TimedProxyUGI getTimedProxyUGI(String user, SegmentTransactionId session)
throws IOException {
    +
    +        Integer segmentId = session.getSegmentId();
    +        synchronized (delayQueues[segmentId]) {
    +            // use the opportunity to cleanup any expired entries
    +            cleanup(segmentId);
    +            TimedProxyUGI timedProxyUGI = cache.get(session);
    +            if (timedProxyUGI == null) {
    +                LOG.info(session.toString() + " Creating proxy user = " + user);
    +                UserGroupInformation proxyUGI =
    +                        UserGroupInformation.createProxyUser(user, UserGroupInformation.getLoginUser());
    +                timedProxyUGI = new TimedProxyUGI(proxyUGI, session);
    +                delayQueues[segmentId].offer(timedProxyUGI);
    +                cache.put(session, timedProxyUGI);
    +            }
    +            timedProxyUGI.incrementCounter();
    +            return timedProxyUGI;
    +        }
    +    }
    +
    +    private cleanup(Integer segmentId) {
    +        // poll segment expiration queue for all expired entries and clean them if possible
  
    +        TimedProxyUGI ugi = null;
    +        while ((ugi = delayQueues[segmentId].poll()) != null) {
    +            // place it back in the queue if still in use and was not closed
    +            if (!closeUGI(ugi)) {
    +                delayQueues[segmentId].offer(ugi);
    +            }
    +            LOG.info("Delay Queue Size for segment " +
    +                    segmentId + " = " + delayQueues[segmentId].size());
    +        }
    +    }
    +    
    +    public void release(TimedProxyUGI timedProxyUGI, boolean forceClean) {
    +
    +        Integer segmentId = timedProxyUGI.getSession().getSegmentId();
    +        
    +        timedProxyUGI.resetTime();
    +        timedProxyUGI.decrementCounter();
    +        
    +        if (forceClean) {
    +            synchronized (delayQueues[segmentId]) {
    +                timedProxyUGI.setExpired();
    +                closeUGI(timedProxyUGI);
    +            }
    +        }
    +    }
    +
    +    private static boolean closeUGI(TimedProxyUGI expiredProxyUGI) {
    +
    +        SegmentTransactionId session = expiredProxyUGI.getSession();
    +        Integer segmentId = session.getSegmentId();
    +        //synchronized (delayQueues[segmentId]) {
    +            String fsMsg = "FileSystem for proxy user = " + expiredProxyUGI.getProxyUGI().getUserName();
    +            try {
    +                // The UGI object is still being used by another thread
    +                if (expiredProxyUGI.inProgress.get() != 0) {
    +                    LOG.info(session.toString() + " Skipping close of " + fsMsg);
    +                    expiredProxyUGI.resetTime(); // update time so that it doesn't expire
until release updates time again
    +                    return false;
    +                }
    +                // expired UGI can be cleaned, as it is not used
    +                // determine if it can be removed from cache also
    +                TimedProxyUGI cachedUGI = cache.get(session);
    +
    +                if (expiredProxyUGI == cachedUGI) {
    +                    // remove it from cache, as cache now has expired entry which is
not in progress
    +                    cache.remove(session);
    +                }
    +                
    +                if (!expiredProxyUGI.isCleaned()) {
    +                    LOG.info(session.toString() + " Closing " + fsMsg +
    +                            " (Cache Size = " + cache.size() + ")");
    +                    FileSystem.closeAllForUGI(expiredProxyUGI.getProxyUGI());
    +                    expiredProxyUGI.setCleaned();
    +                }
    +                
    --- End diff --
    
    Should we be returning `true` here?


---

Mime
View raw message