text:
 const functions = require("@google-cloud/functions-framework");
 const { Storage } = require('@google-cloud/storage');
 const { BigQuery } = require('@google-cloud/bigquery');
 const bunyan = require('bunyan');
 const schema = require('./schema.json');

 const { LoggingBunyan } = require('@google-cloud/logging-bunyan');

 const loggingBunyan = new LoggingBunyan();

 const logger = bunyan.createLogger({
   name: 'gcs2bq',
   streams: [
     { stream: process.stdout, level: 'info' },
     loggingBunyan.stream('info')
   ],
 });

 const DATASET = process.env.DATASET
 const TABLE = process.env.TABLE
 const LOCATION = process.env.LOCATION

 async function loadCSVFromGCS(datasetId, tableId, timePartition, bucket, filename) {
   const metadata = {
     sourceFormat: 'CSV',
     skipLeadingRows: 1,
     maxBadRecords: 1000,
     schema: {
       fields: schema
     },
    location: LOCATION
   };

   logger.info(`Trying to load ${bucket}/${filename} in ${timePartition} time partition of table ${tableId}...`);
   const bigquery = new BigQuery();
   const storage = new Storage();
   const [job] = await bigquery
     .dataset(datasetId)
     .table(`${tableId}\$${timePartition}`)
     .load(storage.bucket(bucket).file(filename), metadata);
   logger.info(`Job ${job.id} completed.`);
   const errors = job.status.errors;
   if (errors && errors.length > 0) {
     logger.info('Errors occurred:' + JSON.stringify(errors));
     throw new Error('File could not be loaded in time partition');
   }
 }

 functions.cloudEvent("gcs2bq", async (cloudEvent) => {

   const data = JSON.parse(Buffer.from(cloudEvent.data.message.data, 'base64').toString());
   logger.info('Notification received');
   logger.info(data);
   const pattern = /([^/]+\/)?[0-9]{14}\-[0-9a-z]{8}\-[0-9a-z]{4}\-[0-9a-z]{4}\-[0-9a-z]{4}\-[0-9a-z]{12}\-api\-from\-([0-9]{8})0000\-to\-([0-9]{8})0000\/result\-[0-9]+\.csv\.gz/
   const result = data.name.match(pattern);
   const timePartition = result[2];

   await loadCSVFromGCS(DATASET, TABLE, timePartition, data.bucket, data.name)

 });


text:
package com.qihoo360.loader2;

import android.app.Service;
import android.content.BroadcastReceiver;
import android.content.ComponentName;
import android.content.Context;
import android.content.Intent;
import android.content.IntentFilter;
import android.content.pm.ProviderInfo;
import android.content.pm.ServiceInfo;
import android.os.IBinder;
import android.os.Parcelable;
import android.os.RemoteException;
import android.support.v4.content.LocalBroadcastManager;
import android.text.TextUtils;

import com.qihoo360.i.Factory;
import com.qihoo360.i.IModule;
import com.qihoo360.i.IPluginManager;
import com.qihoo360.mobilesafe.api.Tasks;
import com.qihoo360.replugin.IHostBinderFetcher;
import com.qihoo360.replugin.RePlugin;
import com.qihoo360.replugin.RePluginConstants;
import com.qihoo360.replugin.RePluginInternal;
import com.qihoo360.replugin.base.IPC;
import com.qihoo360.replugin.component.activity.DynamicClassProxyActivity;
import com.qihoo360.replugin.component.dummy.DummyActivity;
import com.qihoo360.replugin.component.dummy.DummyProvider;
import com.qihoo360.replugin.component.dummy.DummyService;
import com.qihoo360.replugin.component.process.PluginProcessHost;
import com.qihoo360.replugin.component.service.server.PluginPitService;
import com.qihoo360.replugin.helper.HostConfigHelper;
import com.qihoo360.replugin.helper.LogDebug;
import com.qihoo360.replugin.helper.LogRelease;
import com.qihoo360.replugin.model.PluginInfo;
import com.qihoo360.replugin.packages.PluginManagerProxy;
import com.qihoo360.replugin.utils.ReflectUtils;

import java.io.FileDescriptor;
import java.io.PrintWriter;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import static com.qihoo360.replugin.helper.LogDebug.LOG;
import static com.qihoo360.replugin.helper.LogDebug.PLUGIN_TAG;
import static com.qihoo360.replugin.helper.LogRelease.LOGR;
import static com.qihoo360.replugin.packages.PluginInfoUpdater.ACTION_UNINSTALL_PLUGIN;

/**
 * @author RePlugin Team
 */
class PmBase {

    private static final String TAG = "PmBase";

    static final String ACTION_NEW_PLUGIN = "ACTION_NEW_PLUGIN";

    static final String CONTAINER_SERVICE_PART = ".loader.s.Service";

    private static final String CONTAINER_PROVIDER_PART = ".loader.p.Provider";

    /**
     *
     */
    private final Context mContext;

    /**
     *
     */
    private final HashSet<String> mContainerActivities = new HashSet<String>();

    /**
     *
     */
    private final HashSet<String> mContainerProviders = new HashSet<String>();

    /**
     *
     */
    private final HashSet<String> mContainerServices = new HashSet<String>();

    /**
     *
     */
    private final HashMap<String, HashMap<String, IModule>> mBuiltinModules = new HashMap<String, HashMap<String, IModule>>();

    /**
     *
     */
    private ClassLoader mClassLoader;

    /**
     * 所有插件
     */
    private final Map<String, Plugin> mPlugins = new ConcurrentHashMap<>();

    /**
     * 仿插件对象,用来实现主程序提供binder给其他模块
     */
    private final HashMap<String, IHostBinderFetcher> mBuiltinPlugins = new HashMap<String, IHostBinderFetcher>();

    /**
     * 动态的类查找表
     */
    private final HashMap<String, DynamicClass> mDynamicClasses = new HashMap<String, DynamicClass>();

    /**
     *
     */
    private String mDefaultPluginName;

    /**
     *
     */
    private Plugin mDefaultPlugin;

    /**
     * TODO init
     */
    long mLocalCookie;

    /**
     * TODO init
     */
    private boolean mNeedRestart;

    /**
     *
     */
    Builder.PxAll mAll;

    /**
     *
     */
    private PmHostSvc mHostSvc;

    /**
     *
     */
    PluginProcessPer mClient;

    /**
     *
     */
    PluginCommImpl mLocal;

    /**
     *
     */
    PluginLibraryInternalProxy mInternal;

    /**
     * insertNewPlugin 时使用的线程锁
     */
    private static final byte[] LOCKER = new byte[0];

    /**
     * 类映射
     */
    private static class DynamicClass {

        String plugin;

        /** @deprecated */
        String classType; // activity, service, provider

        Class defClass;

        String className;
    }

    static final void cleanIntentPluginParams(Intent intent) {
        // 防止 intent 攻击
        try {
            intent.removeExtra(IPluginManager.KEY_COMPATIBLE);
            intent.removeExtra(IPluginManager.KEY_PLUGIN);
            intent.removeExtra(IPluginManager.KEY_ACTIVITY);
        } catch (Exception e) {
            // ignore
        }
    }

    PmBase(Context context) {
        //
        mContext = context;

        // TODO init
        //init(context, this);

        if (PluginManager.sPluginProcessIndex == IPluginManager.PROCESS_UI || PluginManager.isPluginProcess()) {
            String suffix;
            if (PluginManager.sPluginProcessIndex == IPluginManager.PROCESS_UI) {
                suffix = "N1";
            } else {
                suffix = "" + PluginManager.sPluginProcessIndex;
            }
            //
            mContainerProviders.add(IPC.getPackageName() + CONTAINER_PROVIDER_PART + suffix);
            //
            mContainerServices.add(IPC.getPackageName() + CONTAINER_SERVICE_PART + suffix);
        }

        //
        mClient = new PluginProcessPer(context, this, PluginManager.sPluginProcessIndex, mContainerActivities);

        //
        mLocal = new PluginCommImpl(context, this);

        //
        mInternal = new PluginLibraryInternalProxy(this);
    }

    void init() {

        RePlugin.getConfig().getCallbacks().initPnPluginOverride();

        if (HostConfigHelper.PERSISTENT_ENABLE) {
            // (默认)"常驻进程"作为插件管理进程,则常驻进程作为Server,其余进程作为Client
            if (IPC.isPersistentProcess()) {
                // 初始化"Server"所做工作
                initForServer();
            } else {
                // 连接到Server
                initForClient();
            }
        } else {
            // "UI进程"作为插件管理进程(唯一进程),则UI进程既可以作为Server也可以作为Client
            if (IPC.isUIProcess()) {
                // 1. 尝试初始化Server所做工作,
                initForServer();

                // 2. 注册该进程信息到"插件管理进程"中
                // 注意:这里无需再做 initForClient,因为不需要再走一次Binder
                PMF.sPluginMgr.attach();

            } else {
                // 其它进程?直接连接到Server即可
                initForClient();
            }
        }

        // 最新快照
        PluginTable.initPlugins(mPlugins);

        // 输出
        if (LOG) {
            for (Plugin p : mPlugins.values()) {
                LogDebug.d(PLUGIN_TAG, "plugin: p=" + p.mInfo);
            }
        }
    }

    /**
     * Persistent(常驻)进程的初始化
     *
     */
    private final void initForServer() {
        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "search plugins from file system");
        }

        mHostSvc = new PmHostSvc(mContext, this);
        PluginProcessMain.installHost(mHostSvc);
        PluginProcessMain.schedulePluginProcessLoop(PluginProcessMain.CHECK_STAGE1_DELAY);

        // 兼容即将废弃的p-n方案 by Jiongxuan Zhang
        mAll = new Builder.PxAll();
        Builder.builder(mContext, mAll);
        refreshPluginMap(mAll.getPlugins());

        // [Newest!] 使用全新的RePlugin APK方案
        // Added by Jiongxuan Zhang
        try {
            List<PluginInfo> l = PluginManagerProxy.load();
            if (l != null) {
                // 将"纯APK"插件信息并入总的插件信息表中,方便查询
                // 这里有可能会覆盖之前在p-n中加入的信息。本来我们就想这么干,以"纯APK"插件为准
                refreshPluginMap(l);
            }
        } catch (RemoteException e) {
            if (LOGR) {
                LogRelease.e(PLUGIN_TAG, "lst.p: " + e.getMessage(), e);
            }
        }
    }

    /**
     * Client(UI进程)的初始化
     *
     */
    private final void initForClient() {
        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "list plugins from persistent process");
        }

        // 1. 先尝试连接
        PluginProcessMain.connectToHostSvc();

        // 2. 然后从常驻进程获取插件列表
        refreshPluginsFromHostSvc();
    }

    /**
     * 从HostSvc(插件管理所在进程)获取所有的插件信息
     */
    private void refreshPluginsFromHostSvc() {
        List<PluginInfo> plugins = null;
        try {
            plugins = PluginProcessMain.getPluginHost().listPlugins();
        } catch (Throwable e) {
            if (LOGR) {
                LogRelease.e(PLUGIN_TAG, "lst.p: " + e.getMessage(), e);
            }
        }

        // 判断是否有需要更新的插件
        // FIXME 执行此操作前,判断下当前插件的运行进程,具体可以限制仅允许该插件运行在一个进程且为自身进程中
        List<PluginInfo> updatedPlugins = null;
        if (isNeedToUpdate(plugins)) {
            if (LOG) {
                LogDebug.d(PLUGIN_TAG, "plugins need to perform update operations");
            }
            try {
                updatedPlugins = PluginManagerProxy.updateAllPlugins();
            } catch (RemoteException e) {
                e.printStackTrace();
            }
        }

        if (updatedPlugins != null) {
            refreshPluginMap(updatedPlugins);
        } else {
            refreshPluginMap(plugins);
        }
    }

    /**
     * 判断列表中是否有需要更新的插件
     *
     * @param plugins 要检查的插件的列表
     * @return 是否有需要更新的插件
     */
    private final boolean isNeedToUpdate(List<PluginInfo> plugins) {
        if (plugins != null) {
            for (PluginInfo info : plugins) {
                if (info.getJSON().optJSONObject("upinfo") != null) {
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * 更新所有的插件信息
     *
     * @param plugins
     */
    private final void refreshPluginMap(List<PluginInfo> plugins) {
        if (plugins == null) {
            return;
        }
        for (PluginInfo info : plugins) {
            Plugin plugin = Plugin.build(info);
            putPluginObject(info, plugin);
        }
    }

    /**
     * 把插件Add到插件列表
     *
     * @param info   待add插件的PluginInfo对象
     * @param plugin 待add插件的Plugin对象
     */
    private void putPluginObject(PluginInfo info, Plugin plugin) {
        if (mPlugins.containsKey(info.getAlias()) || mPlugins.containsKey(info.getPackageName())) {
            if (LOG) {
                LogDebug.d(PLUGIN_TAG, "当前内置插件列表中已经有" + info.getName() + ",需要看看谁的版本号大。");
            }

            // 找到已经存在的
            Plugin existedPlugin = mPlugins.get(info.getPackageName());
            if (existedPlugin == null) {
                existedPlugin = mPlugins.get(info.getAlias());
            }

            if (existedPlugin.mInfo.getVersion() < info.getVersion()) {
                if (LOG) {
                    LogDebug.d(PLUGIN_TAG, "新传入的纯APK插件, name=" + info.getName() + ", 版本号比较大,ver=" + info.getVersion() + ",以TA为准。");
                }

                // 同时加入PackageName和Alias(如有)
                mPlugins.put(info.getPackageName(), plugin);
                if (!TextUtils.isEmpty(info.getAlias())) {
                    // 即便Alias和包名相同也可以再Put一次,反正只是覆盖了相同Value而已
                    mPlugins.put(info.getAlias(), plugin);
                }
            } else {
                if (LOG) {
                    LogDebug.d(PLUGIN_TAG, "新传入的纯APK插件" + info.getName() + "版本号还没有内置的大,什么都不做。");
                }
            }
        } else {
            // 同时加入PackageName和Alias(如有)
            mPlugins.put(info.getPackageName(), plugin);
            if (!TextUtils.isEmpty(info.getAlias())) {
                // 即便Alias和包名相同也可以再Put一次,反正只是覆盖了相同Value而已
                mPlugins.put(info.getAlias(), plugin);
            }
        }
    }

    final void attach() {
        //
        try {
            mDefaultPluginName = PluginProcessMain.getPluginHost().attachPluginProcess(IPC.getCurrentProcessName(), PluginManager.sPluginProcessIndex, mClient, mDefaultPluginName);
        } catch (Throwable e) {
            if (LOGR) {
                LogRelease.e(PLUGIN_TAG, "c.n.a: " + e.getMessage(), e);
            }
        }
    }

    final void installBuiltinPlugin(String name, IHostBinderFetcher p) {
        synchronized (mBuiltinPlugins) {
            mBuiltinPlugins.put(name, p);
        }
    }

    final void callAttach() {
        //
        mClassLoader = PmBase.class.getClassLoader();

        // 挂载
        for (Plugin p : mPlugins.values()) {
            p.attach(mContext, mClassLoader, mLocal);
        }

        // 加载默认插件
        if (PluginManager.isPluginProcess()) {
            if (!TextUtils.isEmpty(mDefaultPluginName)) {
                //
                Plugin p = mPlugins.get(mDefaultPluginName);
                if (p != null) {
                    boolean rc = p.load(Plugin.LOAD_APP, true);
                    if (!rc) {
                        if (LOG) {
                            LogDebug.d(PLUGIN_TAG, "failed to load default plugin=" + mDefaultPluginName);
                        }
                    }
                    if (rc) {
                        mDefaultPlugin = p;
                        mClient.init(p);
                    }
                }
            }
        }
    }

    /**
     * @param name
     * @param modc
     * @param module
     */
    final void addBuiltinModule(String name, Class<? extends IModule> modc, IModule module) {
        HashMap<String, IModule> modules = mBuiltinModules.get(name);
        if (modules == null) {
            modules = new HashMap<String, IModule>();
            mBuiltinModules.put(name, modules);
        }
        modules.put(modc.getName(), module);
    }

    final boolean addDynamicClass(String className, String plugin, String type, String target, Class defClass) {
        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "addDynamicClass: class=" + className + " plugin=" + plugin + " type=" + type + " target=" + target + " def=" + defClass);
        }
        if (mDynamicClasses.containsKey(className)) {
            return false;
        }
        DynamicClass dc = new DynamicClass();
        dc.plugin = plugin;
        dc.classType = type;
        dc.className = target;
        dc.defClass = defClass;
        mDynamicClasses.put(className, dc);
        return true;
    }

    /**
     * 检查插件的某个类是否是动态注册的
     *
     * @param plugin    插件名称
     * @param className 要动态注册的类
     * @return 插件的这个类是否是动态类
     */
    final boolean isDynamicClass(String plugin, String className) {
        if (!TextUtils.isEmpty(className) && !TextUtils.isEmpty(plugin)) {
            DynamicClass dc = mDynamicClasses.get(className);
            if (dc != null) {
                return plugin.equals(dc.plugin);
            }
        }
        return false;
    }

    final void removeDynamicClass(String className) {
        mDynamicClasses.remove(className);
    }

    /**
     * 返回 className 对应的 插件名称
     *
     * @param className 插件名称
     * @return 返回动态注册类对应的插件名称
     */
    final String getPluginByDynamicClass(String className) {
        DynamicClass dc = mDynamicClasses.get(className);
        if (dc != null) {
            return dc.plugin;
        }
        return "";
    }

    final void callAppCreate() {
        // 计算/获取cookie
        if (IPC.isPersistentProcess()) {
            mLocalCookie = PluginProcessMain.getPersistentCookie();
        } else {
//            try {
//                mLocalCookie = PmCore.fetchPersistentCookie();
//            } catch (RuntimeException e) {
//                //
//                LogDebug.i(PLUGIN_TAG, "catch exception: " + e.getMessage(), e);
//                //
//                String processName = mContext.getApplicationInfo().packageName + MobileSafeApplication.PERSIST_PROCESS_POSFIX;
//                int uid = mContext.getApplicationInfo().uid;
//                int flags = 0;
//                List<ProviderInfo> providers = mContext.getPackageManager().queryContentProviders(processName, uid, flags);
//                LogDebug.i(PLUGIN_TAG, "providers.size=" + (providers != null ? providers.size() : "null"));
//                if (providers != null) {
//                    for (ProviderInfo pi : providers) {
//                        LogDebug.i(PLUGIN_TAG, "name=" + pi.name + " auth=" + pi.authority);
//                    }
//                }
//                //
//                throw e;
//            }
        }
        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "initial local cookie=" + mLocalCookie);
        }

//        // 退出监控
//        if (IPC.isPersistentProcess()) {
//            // 异步通知,否则可能发生如下错误:
//            //Attempt to invoke virtual method 'android.os.Looper android.content.Context.getMainLooper()' on a null object reference
//            //java.lang.NullPointerException: Attempt to invoke virtual method 'android.os.Looper android.content.Context.getMainLooper()' on a null object reference
//            //    at android.os.Parcel.readException(Parcel.java:1546)
//            //    at android.os.Parcel.readException(Parcel.java:1493)
//            //    at com.qihoo360.loader2.IPluginClient$Stub$Proxy.sendIntent(IPluginClient.java:165)
//            //    at com.qihoo360.loader2.PmCore.sendIntent2Process(PmCore.java:386)
//            //    at com.qihoo360.loader2.PluginManager.sendIntent2Process(PluginManager.java:1124)
//            //    at com.qihoo360.loader2.MP.sendLocalBroadcast2All(MP.java:83)
//            //    at com.qihoo360.mobilesafe.ui.index.IPC.sendLocalBroadcast2All(IPC.java:196)
//            //    at com.qihoo360.mobilesafe.api.IPC.sendLocalBroadcast2All(IPC.java:131)
//            Tasks.post2UI(new Runnable() {
//
//                @Override
//                public void run() {
//                    Intent intent = new Intent(ACTION_PERSISTENT_NEW_COOKIE);
//                    intent.putExtra(KEY_COOKIE, mLocalCookie);
//                    IPC.sendLocalBroadcast2All(mContext, intent);
//                }
//            });
//        }
//
//        if (sPluginProcessIndex >= 0 && sPluginProcessIndex < Constant.STUB_PROCESS_COUNT) {
//            IntentFilter filter = new IntentFilter(ACTION_PERSISTENT_NEW_COOKIE);
//            LocalBroadcastManager.getInstance(mContext).registerReceiver(new BroadcastReceiver() {
//
//                @Override
//                public void onReceive(Context context, Intent intent) {
//                    if (ACTION_PERSISTENT_NEW_COOKIE.equals(intent.getAction())) {
//                        long cookie = intent.getLongExtra(KEY_COOKIE, 0);
//                        if (LOG) {
//                            LogDebug.d(PLUGIN_TAG, "received cookie=" + cookie);
//                        }
//                        if (mLocalCookie != cookie) {
//                            if (LOG) {
//                                LogDebug.d(PLUGIN_TAG, "received new cookie=" + cookie + " old=" + mLocalCookie + " quit ...");
//                            }
//                            // 退出
//                            System.exit(0);
//                        }
//                    }
//                }
//            }, filter);
//        }

        if (!IPC.isPersistentProcess()) {
            // 由于常驻进程已经在内部做了相关的处理,此处仅需要在UI进程注册并更新即可
            registerReceiverAction(ACTION_NEW_PLUGIN);
            registerReceiverAction(ACTION_UNINSTALL_PLUGIN);
        }
    }

    /**
     * @param action
     */
    private final void registerReceiverAction(final String action) {
        IntentFilter filter = new IntentFilter(action);
        LocalBroadcastManager.getInstance(mContext).registerReceiver(new BroadcastReceiver() {

            @Override
            public void onReceive(Context context, Intent intent) {
                if (action.equals(intent.getAction())) {
                    PluginInfo info = intent.getParcelableExtra("obj");
                    if (info != null) {
                        switch (action) {
                            case ACTION_NEW_PLUGIN:
                                // 非常驻进程上下文
                                newPluginFound(info, intent.getBooleanExtra(RePluginConstants.KEY_PERSIST_NEED_RESTART, false));
                                break;
                            case ACTION_UNINSTALL_PLUGIN:
                                pluginUninstalled(info);
                                break;
                        }
                    }
                }
            }
        }, filter);
    }

    /**
     * @param className
     * @param resolve
     * @return
     */
    final Class<?> loadClass(String className, boolean resolve) {
        // 加载Service中介坑位
        if (className.startsWith(PluginPitService.class.getName())) {
            if (LOG) {
                LogDebug.i(TAG, "loadClass: Loading PitService Class... clz=" + className);
            }
            return PluginPitService.class;
        }

        //
        if (mContainerActivities.contains(className)) {
            Class<?> c = mClient.resolveActivityClass(className);
            if (c != null) {
                return c;
            }
            // 输出warn日志便于查看
            // use DummyActivity orig=
            if (LOGR) {
                LogRelease.w(PLUGIN_TAG, "p m hlc u d a o " + className);
            }
            return DummyActivity.class;
        }

        //
        if (mContainerServices.contains(className)) {
            Class<?> c = loadServiceClass(className);
            if (c != null) {
                return c;
            }
            // 输出warn日志便于查看
            // use DummyService orig=
            if (LOGR) {
                LogRelease.w(PLUGIN_TAG, "p m hlc u d s o " + className);
            }
            return DummyService.class;
        }

        //
        if (mContainerProviders.contains(className)) {
            Class<?> c = loadProviderClass(className);
            if (c != null) {
                return c;
            }
            // 输出warn日志便于查看
            // use DummyProvider orig=
            if (LOGR) {
                LogRelease.w(PLUGIN_TAG, "p m hlc u d p o " + className);
            }
            return DummyProvider.class;
        }

        // 插件定制表
        DynamicClass dc = mDynamicClasses.get(className);
        if (dc != null) {
            final Context context = RePluginInternal.getAppContext();
            PluginDesc desc = PluginDesc.get(dc.plugin);

            if (LOG) {
                LogDebug.d("loadClass", "desc=" + desc);
                if (desc != null) {
                    LogDebug.d("loadClass", "desc.isLarge()=" + desc.isLarge());
                }
                LogDebug.d("loadClass", "RePlugin.isPluginDexExtracted(" + dc.plugin + ") = " + RePlugin.isPluginDexExtracted(dc.plugin));
            }

            // 加载动态类时,如果其对应的插件未下载,则转到代理类
            if (desc != null) {
                String plugin = desc.getPluginName();
                if (PluginTable.getPluginInfo(plugin) == null) {
                    if (LOG) {
                        LogDebug.d("loadClass", "plugin=" + plugin + " not found, return DynamicClassProxyActivity.class");
                    }
                    return DynamicClassProxyActivity.class;
                }
            }

            /* 加载未安装的大插件时,启动一个过度 Activity */
            // todo fixme 仅对 activity 类型才弹窗
            boolean needStartLoadingActivity = (desc != null && desc.isLarge() && !RePlugin.isPluginDexExtracted(dc.plugin));
            if (LOG) {
                LogDebug.d("loadClass", "needStartLoadingActivity = " + needStartLoadingActivity);
            }
            if (needStartLoadingActivity) {
                Intent intent = new Intent();
                intent.setFlags(Intent.FLAG_ACTIVITY_NEW_TASK);
                // fixme 将 PluginLoadingActivity2 移到 replugin 中来,不写死
                intent.setComponent(new ComponentName(IPC.getPackageName(), "com.qihoo360.loader2.updater.PluginLoadingActivity2"));
                context.startActivity(intent);
            }

            Plugin p = loadAppPlugin(dc.plugin);
            if (LOG) {
                LogDebug.d("loadClass", "p=" + p);
            }
            if (p != null) {
                try {
                    Class<?> cls = p.getClassLoader().loadClass(dc.className);
                    if (needStartLoadingActivity) {
                        // 发广播给过度 Activity,让其关闭
                        // fixme 发送给 UI 进程
                        Tasks.postDelayed2Thread(new Runnable() {
                            @Override
                            public void run() {
                                if (LOG) {
                                    LogDebug.d("loadClass", "发广播,让 PluginLoadingActivity2 消失");
                                }
                                IPC.sendLocalBroadcast2All(context, new Intent("com.qihoo360.replugin.load_large_plugin.dismiss_dlg"));
                            }
                        }, 300);
                        // IPC.sendLocalBroadcast2Process(context, IPC.getPersistentProcessName(), new Intent("com.qihoo360.replugin.load_large_plugin.dismiss_dlg"), )
                    }
                    return cls;
                } catch (Throwable e) {
                    if (LOGR) {
                        LogRelease.w(PLUGIN_TAG, "p m hlc dc " + className, e);
                    }
                }
            } else {
                if (LOG) {
                    LogDebug.d("loadClass", "加载 " + dc.plugin + " 失败");
                }
                Tasks.postDelayed2Thread(new Runnable() {
                    @Override
                    public void run() {
                        IPC.sendLocalBroadcast2All(context, new Intent("com.qihoo360.replugin.load_large_plugin.dismiss_dlg"));
                    }
                }, 300);
            }
            if (LOGR) {
                LogRelease.w(PLUGIN_TAG, "p m hlc dc failed: " + className + " t=" + dc.className + " tp=" + dc.classType + " df=" + dc.defClass);
            }
            // return dummy class
            if ("activity".equals(dc.classType)) {
                return DummyActivity.class;
            } else if ("service".equals(dc.classType)) {
                return DummyService.class;
            } else if ("provider".equals(dc.classType)) {
                return DummyProvider.class;
            }
            return dc.defClass;
        }

        //
        return loadDefaultClass(className);
    }

    /**
     * @param className
     * @return
     */
    private final Class<?> loadServiceClass(String className) {
        //
        Plugin p = mDefaultPlugin;
        if (p == null) {
            if (LOG) {
                LogDebug.d(PLUGIN_TAG, "plugin service loader: not found default plugin,  in=" + className);
            }
            return null;
        }

        ServiceInfo services[] = p.mLoader.mPackageInfo.services;
        if (services == null || services.length <= 0) {
            if (LOG) {
                LogDebug.d(PLUGIN_TAG, "plugin service loader: manifest not item found");
            }
            return null;
        }

        String service = services[0].name;

        ClassLoader cl = p.getClassLoader();
        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "plugin service loader: in=" + className + " target=" + service);
        }
        Class<?> c = null;
        try {
            c = cl.loadClass(service);
        } catch (Throwable e) {
            if (LOG) {
                LogDebug.d(PLUGIN_TAG, e.getMessage(), e);
            }
        }
        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "plugin service loader: c=" + c + ", loader=" + cl);
        }
        return c;
    }

    /**
     * @param className
     * @return
     */
    private final Class<?> loadProviderClass(String className) {
        //
        Plugin p = mDefaultPlugin;
        if (p == null) {
            if (LOG) {
                LogDebug.d(PLUGIN_TAG, "plugin provider loader: not found default plugin,  in=" + className);
            }
            return null;
        }

        ProviderInfo[] providers = p.mLoader.mPackageInfo.providers;
        if (providers == null || providers.length <= 0) {
            if (LOG) {
                LogDebug.d(PLUGIN_TAG, "plugin provider loader: manifest not item found");
            }
            return null;
        }

        String provider = providers[0].name;

        ClassLoader cl = p.getClassLoader();
        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "plugin provider loader: in=" + className + " target=" + provider);
        }
        Class<?> c = null;
        try {
            c = cl.loadClass(provider);
        } catch (Throwable e) {
            if (LOG) {
                LogDebug.d(PLUGIN_TAG, e.getMessage(), e);
            }
        }
        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "plugin provider loader: c=" + c + ", loader=" + cl);
        }
        return c;
    }

    /**
     * @param className
     * @return
     */
    private final Class<?> loadDefaultClass(String className) {
        //
        Plugin p = mDefaultPlugin;
        if (p == null) {
            if (PluginManager.isPluginProcess()) {
                if (LOG) {
                    LogDebug.d(PLUGIN_TAG, "plugin class loader: not found default plugin,  in=" + className);
                }
            }
            return null;
        }

        ClassLoader cl = p.getClassLoader();
        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "plugin class loader: in=" + className);
        }
        Class<?> c = null;
        try {
            c = cl.loadClass(className);
        } catch (Throwable e) {
            if (LOG) {
                if (e != null && e.getCause() instanceof ClassNotFoundException) {
                    if (LOG) {
                        LogDebug.d(PLUGIN_TAG, "plugin classloader not found className=" + className);
                    }
                } else {
                    if (LOG) {
                        LogDebug.d(PLUGIN_TAG, e.getMessage(), e);
                    }
                }
            }
        }
        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "plugin class loader: c=" + c + ", loader=" + cl);
        }
        return c;
    }

    void dump(FileDescriptor fd, PrintWriter writer, String[] args) {
        if (RePluginInternal.FOR_DEV) {
            // 是否加载插件指令
            {
                boolean load = false;
                for (String a : args) {
                    if (load) {
                        Context c = Factory.queryPluginContext(a);
                        writer.println("plugin.c=" + c);
                        return;
                    }
                    if (a.equals("--load")) {
                        load = true;
                    }
                }
            }

            // 是否启动插件进程
            {
                boolean load = false;
                for (String a : args) {
                    if (load) {
                        try {
                            PluginBinderInfo info = new PluginBinderInfo(PluginBinderInfo.BINDER_REQUEST);
                            /*IPluginClient client = */MP.startPluginProcess(a, IPluginManager.PROCESS_AUTO, info);
                        } catch (Throwable e) {
                            e.printStackTrace();
                        }
                        return;
                    }
                    if (a.equals("--start-plugin-process")) {
                        load = true;
                    }
                }
            }

            // dump原因
            {
                for (String a : args) {
                    if (a.equals("--reason")) {
                        writer.println("--- Reason ---");
                        if (Plugin.sLoadedReasons != null) {
                            for (String reason : Plugin.sLoadedReasons) {
                                writer.println(reason);
                            }
                        }
                        return;
                    }
                }
            }

            // dump binder原因
            {
                for (String a : args) {
                    if (a.equals("--binder-reason")) {
                        writer.println("--- Binder Reason ---");
                        if (MP.sBinderReasons != null) {
                            for (String key : MP.sBinderReasons.keySet()) {
                                writer.println("binder: " + key);
                                writer.println(MP.sBinderReasons.get(key));
                            }
                        }
                        return;
                    }
                }
            }

            // 是否启动插件指令
            {
                boolean start = false;
                String plugin = "";
                String activity = "";
                for (String a : args) {
                    if (start) {
                        if (TextUtils.isEmpty(plugin)) {
                            plugin = a;
                            continue;
                        }
                        if (TextUtils.isEmpty(activity)) {
                            activity = a;
                            continue;
                        }
                    }
                    if (a.equals("--start")) {
                        start = true;
                    }
                }
                if (start) {
                    if (!TextUtils.isEmpty(plugin) && !TextUtils.isEmpty(activity)) {
                        Intent intent = new Intent();
                        intent.setFlags(Intent.FLAG_ACTIVITY_NEW_TASK);
                        Factory.startActivity(mContext, intent, plugin, activity, IPluginManager.PROCESS_AUTO);
                    } else {
                        if (LOG) {
                            LogDebug.d(PLUGIN_TAG, "need {plugin} and {activity}");
                        }
                    }
                    return;
                }
            }

            ReflectUtils.dumpObject(this, fd, writer, args);
            writer.println();

            writer.println("--- plugins V2 ---");
            writer.println("--- plugins.size = " + mPlugins.size() + " ---");
            for (Plugin p : mPlugins.values()) {
                writer.println(p.mInfo);
            }
            writer.println();

            PluginProcessMain.dump(fd, writer, args);

            writer.println("--- plugins.cached objects ---");
            Plugin.dump(fd, writer, args);
            writer.println();
        }
    }

    final IBinder getHostBinder() {
        return mHostSvc;
    }

    final boolean isActivity(String name) {
        return mContainerActivities.contains(name);
    }

    final Plugin getPlugin(String plugin) {
        return mPlugins.get(plugin);
    }

    final Plugin loadPackageInfoPlugin(String plugin, PluginCommImpl pm) {
        Plugin p = Plugin.cloneAndReattach(mContext, mPlugins.get(plugin), mClassLoader, pm);
        return loadPlugin(p, Plugin.LOAD_INFO, true);
    }

    final Plugin loadResourcePlugin(String plugin, PluginCommImpl pm) {
        Plugin p = Plugin.cloneAndReattach(mContext, mPlugins.get(plugin), mClassLoader, pm);
        return loadPlugin(p, Plugin.LOAD_RESOURCES, true);
    }

    final Plugin loadDexPlugin(String plugin, PluginCommImpl pm) {
        Plugin p = Plugin.cloneAndReattach(mContext, mPlugins.get(plugin), mClassLoader, pm);
        return loadPlugin(p, Plugin.LOAD_DEX, true);
    }

    final Plugin loadAppPlugin(String plugin) {
        return loadPlugin(mPlugins.get(plugin), Plugin.LOAD_APP, true);
    }

    // 底层接口
    final Plugin loadPlugin(PluginInfo pi, PluginCommImpl pm, int loadType, boolean useCache) {
        Plugin p = Plugin.build(pi);
        p.attach(mContext, mClassLoader, pm);
        return loadPlugin(p, loadType, useCache);
    }

    // 底层接口
    final Plugin loadPlugin(Plugin p, int loadType, boolean useCache) {
        if (p == null) {
            return null;
        }
        if (!p.load(loadType, useCache)) {
            if (LOGR) {
                LogRelease.e(PLUGIN_TAG, "pmb.lp: f to l. lt=" + loadType + "; i=" + p.mInfo);
            }
            return null;
        }
        return p;
    }

    final Plugin lookupPlugin(ClassLoader loader) {
        for (Plugin p : mPlugins.values()) {
            if (p != null && p.getClassLoader() == loader) {
                return p;
            }
        }
        return null;
    }

    final void insertNewPlugin(PluginInfo info) {
        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "insert new plugin: info=" + info);
        }
        synchronized (LOCKER) {

            // 检查插件是否已经被禁用
            if (RePlugin.getConfig().getCallbacks().isPluginBlocked(info)) {
                if (LOG) {
                    LogDebug.d(PLUGIN_TAG, "insert new plugin: plugin is blocked, in=" + info);
                }
                return;
            }

            Plugin p = mPlugins.get(info.getName());

            // 如果是内置插件,新插件extract成功,则直接替换
            // TODO 考虑加锁?
            if (p != null && p.mInfo.getType() == PluginInfo.TYPE_BUILTIN && info.getType() == PluginInfo.TYPE_PN_INSTALLED) {
                // next

            } else if (p != null && p.isInitialized()) {
                // 检查该插件是否已加载
                if (LOG) {
                    LogDebug.d(PLUGIN_TAG, "insert new plugin: failed cause plugin has loaded, plugin=" + info);
                }
                // 设置是否需要重启标志
                mNeedRestart = true;
                return;
            }

            // 此处直接使用该插件,没有考虑是否只采用最新版
            if (LOG) {
                LogDebug.d(PLUGIN_TAG, "insert new plugin: ok: plugin=" + info);
            }
            Plugin plugin = Plugin.build(info);
            plugin.attach(mContext, mClassLoader, mLocal);

            // 同时加入PackageName和Alias(如有)
            putPluginObject(info, plugin);
        }
    }

    final void newPluginFound(PluginInfo info, boolean persistNeedRestart) {
        // 更新最新插件表
        PluginTable.updatePlugin(info);

        // 更新可加载插件表
        insertNewPlugin(info);

        // 清空插件的状态(解禁)
        PluginStatusController.setStatus(info.getName(), info.getVersion(), PluginStatusController.STATUS_OK);

        if (IPC.isPersistentProcess()) {
            persistNeedRestart = mNeedRestart;
        }

        // 输出一个日志
        if (LOGR) {
            LogRelease.i(PLUGIN_TAG, "p.m. n p f n=" + info.getName() + " b1=" + persistNeedRestart + " b2=" + mNeedRestart);
        }

        // 通知本进程:通知给外部使用者
        Intent intent = new Intent(RePluginConstants.ACTION_NEW_PLUGIN);
        intent.putExtra(RePluginConstants.KEY_PLUGIN_INFO, (Parcelable) info);
        intent.putExtra(RePluginConstants.KEY_PERSIST_NEED_RESTART, persistNeedRestart);
        intent.putExtra(RePluginConstants.KEY_SELF_NEED_RESTART, mNeedRestart);
        LocalBroadcastManager.getInstance(mContext).sendBroadcast(intent);
    }

    final void pluginUninstalled(PluginInfo info) {
        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "Clear plugin cache. pn=" + info.getName());
        }

        // 移除卸载插件的HashMap缓存
        if (mPlugins.containsKey(info.getName())) {
            mPlugins.remove(info.getName());
        }

        // 移除卸载插件表快照
        PluginTable.removeInfo(info);

        // 移除内存中插件的PackageInfo、Resources、ComponentList和DexClassLoader缓存对象
        Plugin.clearCachedPlugin(Plugin.queryCachedFilename(info.getName()));
    }

    final IPluginClient startPluginProcessLocked(String plugin, int process, PluginBinderInfo info) {
        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "start plugin process: plugin=" + plugin + " info=" + info);
        }

        // 强制使用UI进程
        if (Constant.ENABLE_PLUGIN_ACTIVITY_AND_BINDER_RUN_IN_MAIN_UI_PROCESS) {
            if (info.request == PluginBinderInfo.ACTIVITY_REQUEST) {
                if (process == IPluginManager.PROCESS_AUTO) {
                    process = IPluginManager.PROCESS_UI;
                }
            }
            if (info.request == PluginBinderInfo.BINDER_REQUEST) {
                if (process == IPluginManager.PROCESS_AUTO) {
                    process = IPluginManager.PROCESS_UI;
                }
            }
        }

        //
        PluginProcessMain.schedulePluginProcessLoop(PluginProcessMain.CHECK_STAGE1_DELAY);

        // 获取
        IPluginClient client = PluginProcessMain.probePluginClient(plugin, process, info);
        if (client != null) {
            if (LOG) {
                LogDebug.d(PLUGIN_TAG, "start plugin process: probe client ok, already running, plugin=" + plugin + " client=" + client);
            }
            return client;
        }

        // 分配
        int index = IPluginManager.PROCESS_AUTO;
        try {
            index = PluginProcessMain.allocProcess(plugin, process);
            if (LOG) {
                LogDebug.d(PLUGIN_TAG, "start plugin process: alloc process ok, plugin=" + plugin + " index=" + index);
            }
        } catch (Throwable e) {
            if (LOGR) {
                LogRelease.e(PLUGIN_TAG, "a.p.p: " + e.getMessage(), e);
            }
        }
        // 分配的坑位不属于UI、和自定义进程,就返回。
        if (!(index == IPluginManager.PROCESS_UI
                || PluginProcessHost.isCustomPluginProcess(index)
                || PluginManager.isPluginProcess(index))) {
            return null;
        }

        // 启动
        boolean rc = PluginProviderStub.proxyStartPluginProcess(mContext, index);
        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "start plugin process: start process ok, plugin=" + plugin + " index=" + index);
        }
        if (!rc) {
            return null;
        }

        // 再次获取
        client = PluginProcessMain.probePluginClient(plugin, process, info);
        if (client == null) {
            if (LOGR) {
                LogRelease.e(PLUGIN_TAG, "spp pc n");
            }
            return null;
        }

        if (LOG) {
            LogDebug.d(PLUGIN_TAG, "start plugin process: probe client ok, plugin=" + plugin + " index=" + info.index);
        }

        return client;
    }

    final IHostBinderFetcher getBuiltinPlugin(String plugin) {
        synchronized (mBuiltinPlugins) {
            return mBuiltinPlugins.get(plugin);
        }
    }

    final HashMap<String, IModule> getBuiltinModules(String plugin) {
        return mBuiltinModules.get(plugin);
    }

    final void handleServiceCreated(Service service) {
//      int pid = Process.myPid();
        try {
            PluginProcessMain.getPluginHost().regService(PluginManager.sPluginProcessIndex, mDefaultPlugin.mInfo.getName(), service.getClass().getName());
        } catch (Throwable e) {
            if (LOGR) {
                LogRelease.e(PLUGIN_TAG, "r.s: " + e.getMessage(), e);
            }
        }

//      // TODO 设置插件服务类的类加载器吗?
//      Intent intent = service.getIntent();
//      if (intent != null) {
//          if (LOG) {
//              LogDebug.d(PLUGIN_TAG, "set service intent cl=" + service.getClassLoader());
//          }
//          intent.setExtrasClassLoader(service.getClassLoader());
//      }
    }

    final void handleServiceDestroyed(Service service) {
//      int pid = Process.myPid();
        try {
            PluginProcessMain.getPluginHost().unregService(PluginManager.sPluginProcessIndex, mDefaultPlugin.mInfo.getName(), service.getClass().getName());
        } catch (Throwable e) {
            if (LOGR) {
                LogRelease.e(PLUGIN_TAG, "ur.s: " + e.getMessage(), e);
            }
        }
    }
}

redpajama_set_name: RedPajamaGithub


text:
autocommit off;
  create class setA(a set int, b set string, c set)
  insert into setA values({1,2}, {'hi'}, {});

  select a+b from setA;
  select a-b from setA;
  select a*b from setA;
  select b+a from setA;
  select b-a from setA;
  select b*a from setA;
  select c+b from setA;
  select c-b from setA;
  select c*b from setA;
  select b+c from setA;
  select b-c from setA;
  select b*c from setA;
  select a+c from setA;
  select a-c from setA;
  select a*c from setA;
  select c+a from setA;
  select c-a from setA;
  select c*a from setA;
rollback;


text:
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">

        <META http-equiv="Content-Type" content="text/html; charset=UTF-8">
        <TITLE>
            Uses of Class org.encog.util.csv.ReadCSV (Encog Core 3.2.0-SNAPSHOT API)
        </TITLE>

        <META NAME="date" CONTENT="2013-04-17">

        <LINK REL ="stylesheet" TYPE="text/css" HREF="../../../../../stylesheet.css" TITLE="Style">

                            Overview
                            Package
                            Class
                          Use
                            Tree
                            Deprecated
                            Index
                            Help

                     PREV
                     NEXT

                    FRAMES
                     NO FRAMES

                        All Classes

            Uses of Class
org.encog.util.csv.ReadCSV

                    Packages that use ReadCSV

            org.encog.app.analyst.csv.basic

            org.encog.app.analyst.csv.normalize

            org.encog.app.analyst.csv.process

                    Uses of ReadCSV in org.encog.app.analyst.csv.basic

            Methods in org.encog.app.analyst.csv.basic with parameters of type ReadCSV

                     String
            BasicCachedFile.getColumnData(String name,
                    ReadCSV csv)

                          Get the data for a specific column.

                     void
            BasicFile.readHeaders(ReadCSV csv)

                          Read the headers from a CSV file.

            Constructors in org.encog.app.analyst.csv.basic with parameters of type ReadCSV

            LoadedRow(ReadCSV csv)

                          Load a row from the specified CSV file.

            LoadedRow(ReadCSV csv,
                    int extra)

                          Construct a loaded row.

                    Uses of ReadCSV in org.encog.app.analyst.csv.normalize

            Methods in org.encog.app.analyst.csv.normalize with parameters of type ReadCSV

                    static double[]
            AnalystNormalizeCSV.extractFields(EncogAnalyst analyst,
                    CSVHeaders headers,
                    ReadCSV csv,
                    int outputLength,
                    boolean skipOutput)

                          Extract fields from a file into a numeric array for machine learning.

                    Uses of ReadCSV in org.encog.app.analyst.csv.process

            Methods in org.encog.app.analyst.csv.process with parameters of type ReadCSV

                     void
            ProcessExtension.init(ReadCSV csv,
                    int theBackwardWindowSize,
                    int theForwardWindowSize)

                            Overview
                            Package
                            Class
                          Use
                            Tree
                            Deprecated
                            Index
                            Help

                     PREV
                     NEXT

                    FRAMES
                     NO FRAMES

                        All Classes

    Copyright © 2013. All Rights Reserved.


text:
How to Contribute
=================
Do you want to contribute fixes or improvements?

   **AWesome!** *Thank you very much, and let's get started.*

Set up a development environment
--------------------------------
The first thing that you need is a development environment so that you can
run the test suite, update the documentation, and everything else that is
involved in contributing.  The easiest way to do that is to create a virtual
environment for your endeavours::

   $ python3 -mvenv env

Don't worry about writing code against previous versions of Python unless
you you don't have a choice.  That is why we run our tests through `tox`_.
If you don't have a choice, then install `virtualenv`_ to create the
environment instead.  The next step is to install the development tools
that this project uses.  These are listed in *requires/development.txt*::

   $ env/bin/pip install -qr requires/development.txt

At this point, you will have everything that you need to develop at your
disposal.  *setup.py* is the swiss-army knife in your development tool
chest.  It provides the following commands:

**./setup.py nosetests**
   Run the test suite using `nose`_ and generate a nice coverage report.

**./setup.py build_sphinx**
   Generate the documentation using `sphinx`_.

**./setup.py flake8**
   Run `flake8`_ over the code and report style violations.

If any of the preceding commands give you problems, then you will have to
fix them **before** your pull request will be accepted.

Running Tests
-------------
The easiest (and quickest) way to run the test suite is to use the
*nosetests* command.  It will run the test suite against the currently
installed python version and report not only the test result but the
test coverage as well::

   $ ./setup.py nosetests

   running nosetests
   running egg_info
   writing dependency_links to sprockets_dynamodb/dependency_links.txt
   writing top-level names to sprockets_dynamodb/top_level.txt
   writing sprockets_dynamodb/PKG-INFO
   reading manifest file 'sprockets_dynamodb/SOURCES.txt'
   reading manifest template 'MANIFEST.in'
   warning: no previously-included files matching '__pycache__'...
   warning: no previously-included files matching '*.swp' found ...
   writing manifest file 'sprockets_dynamodb/SOURCES.txt'
   ...

   Name                       Stmts   Miss Branch BrMiss  Cover   Missing
   ----------------------------------------------------------------------
   ...
   ----------------------------------------------------------------------
   TOTAL                         95      2     59      2    97%
   ----------------------------------------------------------------------
   Ran 44 tests in 0.054s

   OK

That's the quick way to run tests.  The slightly longer way is to run
the `tox`_ utility.  It will run the test suite against all of the
supported python versions in parallel.  This is essentially what Travis-CI
will do when you issue a pull request anyway::

   $ env/bin/tox
   py27 recreate: /.../sprockets-dynamodb/build/tox/py27
   GLOB sdist-make: /.../sprockets-dynamodb/setup.py
   py34 recreate: /.../sprockets-dynamodb/build/tox/py34
   py27 installdeps: -rtest-requirements.txt, mock
   py34 installdeps: -rtest-requirements.txt
   py27 inst: /.../sprockets-dynamodb/build/tox/dist/sprockets-dynamodb-0.0.0.zip
   py27 runtests: PYTHONHASHSEED='2156646470'
   py27 runtests: commands[0] | /../sprockets-dynamodb/build/tox/py27/bin/nosetests
   py34 inst: /../sprockets-dynamodb/.build/tox/dist/sprockets-dynamodb-0.0.0.zip
   py34 runtests: PYTHONHASHSEED='2156646470'
   py34 runtests: commands[0] | /.../sprockets-dynamodb/build/tox/py34/bin/nosetests
   _________________________________ summary _________________________________
     py27: commands succeeded
     py34: commands succeeded
     congratulations :)

This is what you want to see.  Now you can make your modifications and keep
the tests passing.

Submitting a Pull Request
-------------------------
Once you have made your modifications, gotten all of the tests to pass,
and added any necessary documentation, it is time to contribute back for
posterity.  You've probably already cloned this repository and created a
new branch.  If you haven't, then checkout what you have as a branch and
roll back *master* to where you found it.  Then push your repository up
to github and issue a pull request.  Describe your changes in the request,
if Travis isn't too annoyed someone will review it, and eventually merge
it back.

.. _flake8: http://flake8.readthedocs.org/
.. _nose: http://nose.readthedocs.org/
.. _sphinx: http://sphinx-doc.org/
.. _tox: http://testrun.org/tox/
.. _virtualenv: http://virtualenv.pypa.io/


text:
mod kernel32 {
  extern "system" {
    pub fn SetLastError(err: usize);
    pub fn GetLastError() -> usize;
  }
}

#[cfg(windows)]
pub fn main() {
    unsafe {
        let expected = 1234;
        kernel32::SetLastError(expected);
        let actual = kernel32::GetLastError();
        println!("actual = {}", actual);
        assert_eq!(expected, actual);
    }
}

#[cfg(any(target_os = "android",
          target_os = "bitrig",
          target_os = "cloudabi",
          target_os = "dragonfly",
          target_os = "emscripten",
          target_os = "freebsd",
          target_os = "linux",
          target_os = "macos",
          target_os = "netbsd",
          target_os = "openbsd",
          target_os = "solaris"))]
pub fn main() { }


text:
#region License
//
// Copyright Tony Beveridge 2015. All rights reserved.
// MIT license applies.
//
#endregion
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using FSM;

namespace Quorum.States {

    public class MasterState : BaseQuorumState {

        public override Task<StateResult> OnEntry(IStateMachineContext<IExecutionContext> context) {
            context.ExecutionContext.IsMaster = true;
            EnsureWorkerActive(context.ExecutionContext);
            return Task.FromResult(StateResult.None);
        }

    }
}


text:
GZ_REGISTER_MODEL_PLUGIN(Gyro)

Gyro::Gyro() {}

Gyro::~Gyro() {}

void Gyro::Load(physics::ModelPtr model, sdf::ElementPtr sdf) {
  this->model = model;

  // Parse SDF properties
  link = model->GetLink(sdf->Get<std::string>("link"));
  if (sdf->HasElement("topic")) {
    topic = sdf->Get<std::string>("topic");
  } else {
    topic = "~/"+sdf->GetAttribute("name")->GetAsString();
  }

  std::string axisString = sdf->Get<std::string>("axis");
  if (axisString == "roll") axis = Roll;
  if (axisString == "pitch") axis = Pitch;
  if (axisString == "yaw") axis = Yaw;

  if (sdf->HasElement("units")) {
    radians = sdf->Get<std::string>("units") != "degrees";
  } else {
    radians = true;
  }
  zero = GetAngle();

  gzmsg << "Initializing gyro: " << topic << " link=" << link->GetName()
        << " axis=" << axis << " radians=" << radians << std::endl;

  // Connect to Gazebo transport for messaging
  std::string scoped_name = model->GetWorld()->GetName()+"::"+model->GetScopedName();
  boost::replace_all(scoped_name, "::", "/");
  node = transport::NodePtr(new transport::Node());
  node->Init(scoped_name);
  command_sub = node->Subscribe(topic+"/control", &Gyro::Callback, this);
  pos_pub = node->Advertise<msgs::Float64>(topic+"/position");
  vel_pub = node->Advertise<msgs::Float64>(topic+"/velocity");

  // Connect to the world update event.
  // This will trigger the Update function every Gazebo iteration
  updateConn = event::Events::ConnectWorldUpdateBegin(boost::bind(&Gyro::Update, this, _1));
}

void Gyro::Update(const common::UpdateInfo &info) {
  msgs::Float64 pos_msg, vel_msg;
  pos_msg.set_data(Limit(GetAngle() - zero));
  pos_pub->Publish(pos_msg);
  vel_msg.set_data(GetVelocity());
  vel_pub->Publish(vel_msg);
}

void Gyro::Callback(const msgs::ConstStringPtr &msg) {
  std::string command = msg->data();
  if (command == "reset") {
    zero = GetAngle();
  } else {
    gzerr << "WARNING: Gyro got unknown command '" << command << "'." << std::endl;
  }
}

double Gyro::GetAngle() {
  if (radians) {
    return link->GetWorldCoGPose().rot.GetAsEuler()[axis];
  } else {
    return link->GetWorldCoGPose().rot.GetAsEuler()[axis] * (180.0 / M_PI);
  }
}

double Gyro::GetVelocity() {
  if (radians) {
    return link->GetRelativeAngularVel()[axis];
  } else {
    return link->GetRelativeAngularVel()[axis] * (180.0 / M_PI);
  }
}

double Gyro::Limit(double value) {
  if (radians) {
    while (true) {
      if (value < -M_PI) value += 2*M_PI;
      else if (value > M_PI) value -= 2*M_PI;
      else break;
    }
  } else {
    while (true) {
      if (value < -180) value += 360;
      else if (value > 180) value -= 360;
      else break;
    }
  }
  return value;
}


text:
nuget pack Hazware.Logging.NLog.nuspec
nuget pack Hazware.Logging.log4net.nuspec


text:
/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package klose.effj.concurrency;

import java.util.concurrent.TimeUnit;

// Properly synchronized cooperative thread termination
public class StopThread {

    private static boolean stopRequested = false;

    private static synchronized void requestStop() {
        stopRequested = true;
    }

    private static synchronized boolean stopRequested() {
        return stopRequested;
    }

    public static void main(String[] args)
            throws InterruptedException {
        Thread backgroundThread = new Thread(() -> {
            int i = 0;
            while (!stopRequested()) {
                i++;
            }
        });
        backgroundThread.start();
        TimeUnit.SECONDS.sleep(1);
        requestStop();
    }
}


text:
/*
Commands for controlling an external mysql process.

Some commands are issued as exec'd tools, some are handled by connecting via
the mysql protocol.
*/

package mysqlctl

import (
	"bufio"
	"errors"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"os/exec"
	"path"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"bytes"

	"golang.org/x/net/context"
	"vitess.io/vitess/go/mysql"
	"vitess.io/vitess/go/stats"
	"vitess.io/vitess/go/vt/dbconfigs"
	"vitess.io/vitess/go/vt/dbconnpool"
	vtenv "vitess.io/vitess/go/vt/env"
	"vitess.io/vitess/go/vt/hook"
	"vitess.io/vitess/go/vt/log"
	"vitess.io/vitess/go/vt/mysqlctl/mysqlctlclient"
)

var (
	// DisableActiveReparents is a flag to disable active
	// reparents for safety reasons. It is used in three places:
	// 1. in this file to skip registering the commands.
	// 2. in vtctld so it can be exported to the UI (different
	// package, that's why it's exported). That way we can disable
	// menu items there, using features.
	// 3. prevents the vtworker from updating replication topology
	// after restarting replication after a split clone/diff.
	DisableActiveReparents = flag.Bool("disable_active_reparents", false, "if set, do not allow active reparents. Use this to protect a cluster using external reparents.")

	dbaPoolSize    = flag.Int("dba_pool_size", 20, "Size of the connection pool for dba connections")
	dbaIdleTimeout = flag.Duration("dba_idle_timeout", time.Minute, "Idle timeout for dba connections")
	appPoolSize    = flag.Int("app_pool_size", 40, "Size of the connection pool for app connections")
	appIdleTimeout = flag.Duration("app_idle_timeout", time.Minute, "Idle timeout for app connections")

	poolDynamicHostnameResolution = flag.Duration("pool_hostname_resolve_interval", 0, "if set force an update to all hostnames and reconnect if changed, defaults to 0 (disabled)")

	socketFile        = flag.String("mysqlctl_socket", "", "socket file to use for remote mysqlctl actions (empty for local actions)")
	mycnfTemplateFile = flag.String("mysqlctl_mycnf_template", "", "template file to use for generating the my.cnf file during server init")

	// masterConnectRetry is used in 'SET MASTER' commands
	masterConnectRetry = flag.Duration("master_connect_retry", 10*time.Second, "how long to wait in between slave -> connection attempts. Only precise to the second.")

	dbaMysqlStats      = stats.NewTimings("MysqlDba", "MySQL DBA stats", "operation")
	allprivsMysqlStats = stats.NewTimings("MysqlAllPrivs", "MySQl Stats for all privs", "operation")
	appMysqlStats      = stats.NewTimings("MysqlApp", "MySQL app stats", "operation")
)

// Mysqld is the object that represents a mysqld daemon running on this server.
type Mysqld struct {
	dbcfgs  *dbconfigs.DBConfigs
	dbaPool *dbconnpool.ConnectionPool
	appPool *dbconnpool.ConnectionPool

	// mutex protects the fields below.
	mutex         sync.Mutex
	onTermFuncs   []func()
	cancelWaitCmd chan struct{}
}

// NewMysqld creates a Mysqld object based on the provided configuration
// and connection parameters.
func NewMysqld(dbcfgs *dbconfigs.DBConfigs) *Mysqld {
	result := &Mysqld{
		dbcfgs: dbcfgs,
	}

	// Create and open the connection pool for dba access.
	result.dbaPool = dbconnpool.NewConnectionPool("DbaConnPool", *dbaPoolSize, *dbaIdleTimeout, *poolDynamicHostnameResolution)
	result.dbaPool.Open(dbcfgs.Dba(), dbaMysqlStats)

	// Create and open the connection pool for app access.
	result.appPool = dbconnpool.NewConnectionPool("AppConnPool", *appPoolSize, *appIdleTimeout, *poolDynamicHostnameResolution)
	result.appPool.Open(dbcfgs.AppWithDB(), appMysqlStats)

	return result
}

// RunMysqlUpgrade will run the mysql_upgrade program on the current
// install.  Will be called only when mysqld is running with no
// network and no grant tables.
func (mysqld *Mysqld) RunMysqlUpgrade() error {
	// Execute as remote action on mysqlctld if requested.
	if *socketFile != "" {
		log.Infof("executing Mysqld.RunMysqlUpgrade() remotely via mysqlctld server: %v", *socketFile)
		client, err := mysqlctlclient.New("unix", *socketFile)
		if err != nil {
			return fmt.Errorf("can't dial mysqlctld: %v", err)
		}
		defer client.Close()
		return client.RunMysqlUpgrade(context.TODO())
	}

	// Find mysql_upgrade. If not there, we do nothing.
	dir, err := vtenv.VtMysqlRoot()
	if err != nil {
		log.Warningf("VT_MYSQL_ROOT not set, skipping mysql_upgrade step: %v", err)
		return nil
	}
	name, err := binaryPath(dir, "mysql_upgrade")
	if err != nil {
		log.Warningf("mysql_upgrade binary not present, skipping it: %v", err)
		return nil
	}

	// Since we started mysql with --skip-grant-tables, we should
	// be able to run mysql_upgrade without any valid user or
	// password. However, mysql_upgrade executes a 'flush
	// privileges' right in the middle, and then subsequent
	// commands fail if we don't use valid credentials. So let's
	// use dba credentials.
	params, err := dbconfigs.WithCredentials(mysqld.dbcfgs.Dba())
	if err != nil {
		return err
	}
	defaultsFile, err := mysqld.defaultsExtraFile(params)
	if err != nil {
		return err
	}
	defer os.Remove(defaultsFile)

	// Run the program, if it fails, we fail.  Note in this
	// moment, mysqld is running with no grant tables on the local
	// socket only, so this doesn't need any user or password.
	args := []string{
		// --defaults-file=* must be the first arg.
		"--defaults-file=" + defaultsFile,
		"--force", // Don't complain if it's already been upgraded.
	}
	cmd := exec.Command(name, args...)
	cmd.Env = []string{os.ExpandEnv("LD_LIBRARY_PATH=$VT_MYSQL_ROOT/lib/mysql")}
	out, err := cmd.CombinedOutput()
	log.Infof("mysql_upgrade output: %s", out)
	return err
}

// Start will start the mysql daemon, either by running the
// 'mysqld_start' hook, or by running mysqld_safe in the background.
// If a mysqlctld address is provided in a flag, Start will run
// remotely.  When waiting for mysqld to start, we will use
// the dba user.
func (mysqld *Mysqld) Start(ctx context.Context, cnf *Mycnf, mysqldArgs ...string) error {
	// Execute as remote action on mysqlctld if requested.
	if *socketFile != "" {
		log.Infof("executing Mysqld.Start() remotely via mysqlctld server: %v", *socketFile)
		client, err := mysqlctlclient.New("unix", *socketFile)
		if err != nil {
			return fmt.Errorf("can't dial mysqlctld: %v", err)
		}
		defer client.Close()
		return client.Start(ctx, mysqldArgs...)
	}

	if err := mysqld.startNoWait(ctx, cnf, mysqldArgs...); err != nil {
		return err
	}

	return mysqld.Wait(ctx, cnf)
}

// startNoWait is the internal version of Start, and it doesn't wait.
func (mysqld *Mysqld) startNoWait(ctx context.Context, cnf *Mycnf, mysqldArgs ...string) error {
	var name string
	ts := fmt.Sprintf("Mysqld.Start(%v)", time.Now().Unix())

	// try the mysqld start hook, if any
	switch hr := hook.NewHook("mysqld_start", mysqldArgs).Execute(); hr.ExitStatus {
	case hook.HOOK_SUCCESS:
		// hook exists and worked, we can keep going
		name = "mysqld_start hook"
	case hook.HOOK_DOES_NOT_EXIST:
		// hook doesn't exist, run mysqld_safe ourselves
		log.Infof("%v: No mysqld_start hook, running mysqld_safe directly", ts)
		dir, err := vtenv.VtMysqlRoot()
		if err != nil {
			return err
		}
		name, err = binaryPath(dir, "mysqld_safe")
		if err != nil {
			// The movement to use systemd means that mysqld_safe is not always provided.
			// This should not be considered an issue do not generate a warning.
			log.Infof("%v: trying to launch mysqld instead", err)
			name, err = binaryPath(dir, "mysqld")
			// If this also fails, return an error.
			if err != nil {
				return err
			}
		}
		arg := []string{
			"--defaults-file=" + cnf.path}
		arg = append(arg, mysqldArgs...)
		env := []string{os.ExpandEnv("LD_LIBRARY_PATH=$VT_MYSQL_ROOT/lib/mysql")}

		cmd := exec.Command(name, arg...)
		cmd.Dir = dir
		cmd.Env = env
		log.Infof("%v %#v", ts, cmd)
		stderr, err := cmd.StderrPipe()
		if err != nil {
			return err
		}
		stdout, err := cmd.StdoutPipe()
		if err != nil {
			return err
		}
		go func() {
			scanner := bufio.NewScanner(stderr)
			for scanner.Scan() {
				log.Infof("%v stderr: %v", ts, scanner.Text())
			}
		}()
		go func() {
			scanner := bufio.NewScanner(stdout)
			for scanner.Scan() {
				log.Infof("%v stdout: %v", ts, scanner.Text())
			}
		}()
		err = cmd.Start()
		if err != nil {
			return err
		}

		mysqld.mutex.Lock()
		mysqld.cancelWaitCmd = make(chan struct{})
		go func(cancel <-chan struct{}) {
			// Wait regardless of cancel, so we don't generate defunct processes.
			err := cmd.Wait()
			log.Infof("%v exit: %v", ts, err)

			// The process exited. Trigger OnTerm callbacks, unless we were cancelled.
			select {
			case <-cancel:
			default:
				mysqld.mutex.Lock()
				for _, callback := range mysqld.onTermFuncs {
					go callback()
				}
				mysqld.mutex.Unlock()
			}
		}(mysqld.cancelWaitCmd)
		mysqld.mutex.Unlock()
	default:
		// hook failed, we report error
		return fmt.Errorf("mysqld_start hook failed: %v", hr.String())
	}

	return nil
}

// Wait returns nil when mysqld is up and accepting connections. It
// will use the dba credentials to try to connect. Use wait() with
// different credentials if needed.
func (mysqld *Mysqld) Wait(ctx context.Context, cnf *Mycnf) error {
	params, err := dbconfigs.WithCredentials(mysqld.dbcfgs.Dba())
	if err != nil {
		return err
	}

	return mysqld.wait(ctx, cnf, params)
}

// wait is the internal version of Wait, that takes credentials.
func (mysqld *Mysqld) wait(ctx context.Context, cnf *Mycnf, params *mysql.ConnParams) error {
	log.Infof("Waiting for mysqld socket file (%v) to be ready...", cnf.SocketFile)

	for {
		select {
		case <-ctx.Done():
			return errors.New("deadline exceeded waiting for mysqld socket file to appear: " + cnf.SocketFile)
		default:
		}

		_, statErr := os.Stat(cnf.SocketFile)
		if statErr == nil {
			// Make sure the socket file isn't stale.
			conn, connErr := mysql.Connect(ctx, params)
			if connErr == nil {
				conn.Close()
				return nil
			}
			log.Infof("mysqld socket file exists, but can't connect: %v", connErr)
		} else if !os.IsNotExist(statErr) {
			return fmt.Errorf("can't stat mysqld socket file: %v", statErr)
		}
		time.Sleep(1000 * time.Millisecond)
	}
}

// Shutdown will stop the mysqld daemon that is running in the background.
//
// waitForMysqld: should the function block until mysqld has stopped?
// This can actually take a *long* time if the buffer cache needs to be fully
// flushed - on the order of 20-30 minutes.
//
// If a mysqlctld address is provided in a flag, Shutdown will run remotely.
func (mysqld *Mysqld) Shutdown(ctx context.Context, cnf *Mycnf, waitForMysqld bool) error {
	log.Infof("Mysqld.Shutdown")

	// Execute as remote action on mysqlctld if requested.
	if *socketFile != "" {
		log.Infof("executing Mysqld.Shutdown() remotely via mysqlctld server: %v", *socketFile)
		client, err := mysqlctlclient.New("unix", *socketFile)
		if err != nil {
			return fmt.Errorf("can't dial mysqlctld: %v", err)
		}
		defer client.Close()
		return client.Shutdown(ctx, waitForMysqld)
	}

	// We're shutting down on purpose. We no longer want to be notified when
	// mysqld terminates.
	mysqld.mutex.Lock()
	if mysqld.cancelWaitCmd != nil {
		close(mysqld.cancelWaitCmd)
		mysqld.cancelWaitCmd = nil
	}
	mysqld.mutex.Unlock()

	// possibly mysql is already shutdown, check for a few files first
	_, socketPathErr := os.Stat(cnf.SocketFile)
	_, pidPathErr := os.Stat(cnf.PidFile)
	if os.IsNotExist(socketPathErr) && os.IsNotExist(pidPathErr) {
		log.Warningf("assuming mysqld already shut down - no socket, no pid file found")
		return nil
	}

	// try the mysqld shutdown hook, if any
	h := hook.NewSimpleHook("mysqld_shutdown")
	hr := h.Execute()
	switch hr.ExitStatus {
	case hook.HOOK_SUCCESS:
		// hook exists and worked, we can keep going
	case hook.HOOK_DOES_NOT_EXIST:
		// hook doesn't exist, try mysqladmin
		log.Infof("No mysqld_shutdown hook, running mysqladmin directly")
		dir, err := vtenv.VtMysqlRoot()
		if err != nil {
			return err
		}
		name, err := binaryPath(dir, "mysqladmin")
		if err != nil {
			return err
		}
		params, err := dbconfigs.WithCredentials(mysqld.dbcfgs.Dba())
		if err != nil {
			return err
		}
		cnf, err := mysqld.defaultsExtraFile(params)
		if err != nil {
			return err
		}
		defer os.Remove(cnf)
		args := []string{
			"--defaults-extra-file=" + cnf,
			"--shutdown-timeout=300",
			"--connect-timeout=30",
			"--wait=10",
			"shutdown",
		}
		env := []string{
			os.ExpandEnv("LD_LIBRARY_PATH=$VT_MYSQL_ROOT/lib/mysql"),
		}
		_, _, err = execCmd(name, args, env, dir, nil)
		if err != nil {
			return err
		}
	default:
		// hook failed, we report error
		return fmt.Errorf("mysqld_shutdown hook failed: %v", hr.String())
	}

	// Wait for mysqld to really stop. Use the socket and pid files as a
	// proxy for that since we can't call wait() in a process we
	// didn't start.
	if waitForMysqld {
		log.Infof("Mysqld.Shutdown: waiting for socket file (%v) and pid file (%v) to disappear",
			cnf.SocketFile, cnf.PidFile)

		for {
			select {
			case <-ctx.Done():
				return errors.New("gave up waiting for mysqld to stop")
			default:
			}

			_, socketPathErr = os.Stat(cnf.SocketFile)
			_, pidPathErr = os.Stat(cnf.PidFile)
			if os.IsNotExist(socketPathErr) && os.IsNotExist(pidPathErr) {
				return nil
			}
			time.Sleep(100 * time.Millisecond)
		}
	}
	return nil
}

// execCmd searches the PATH for a command and runs it, logging the output.
// If input is not nil, pipe it to the command's stdin.
func execCmd(name string, args, env []string, dir string, input io.Reader) (cmd *exec.Cmd, output string, err error) {
	cmdPath, _ := exec.LookPath(name)
	log.Infof("execCmd: %v %v %v", name, cmdPath, args)

	cmd = exec.Command(cmdPath, args...)
	cmd.Env = env
	cmd.Dir = dir
	if input != nil {
		cmd.Stdin = input
	}
	out, err := cmd.CombinedOutput()
	output = string(out)
	if err != nil {
		log.Infof("execCmd: %v failed: %v", name, err)
		err = fmt.Errorf("%v: %v, output: %v", name, err, output)
	}
	log.Infof("execCmd: %v output: %v", name, output)
	return cmd, output, err
}

// binaryPath does a limited path lookup for a command,
// searching only within sbin and bin in the given root.
func binaryPath(root, binary string) (string, error) {
	subdirs := []string{"sbin", "bin", "libexec"}
	for _, subdir := range subdirs {
		binPath := path.Join(root, subdir, binary)
		if _, err := os.Stat(binPath); err == nil {
			return binPath, nil
		}
	}
	return "", fmt.Errorf("%s not found in any of %s/{%s}",
		binary, root, strings.Join(subdirs, ","))
}

// InitConfig will create the default directory structure for the mysqld process,
// generate / configure a my.cnf file.
func (mysqld *Mysqld) InitConfig(cnf *Mycnf) error {
	log.Infof("mysqlctl.InitConfig")
	err := mysqld.createDirs(cnf)
	if err != nil {
		log.Errorf("%s", err.Error())
		return err
	}
	root, err := vtenv.VtRoot()
	if err != nil {
		log.Errorf("%s", err.Error())
		return err
	}

	// Set up config files.
	if err = mysqld.initConfig(root, cnf, cnf.path); err != nil {
		log.Errorf("failed creating %v: %v", cnf.path, err)
		return err
	}
	return nil
}

// Init will create the default directory structure for the mysqld process,
// generate / configure a my.cnf file install a skeleton database,
// and apply the provided initial SQL file.
func (mysqld *Mysqld) Init(ctx context.Context, cnf *Mycnf, initDBSQLFile string) error {
	log.Infof("mysqlctl.Init")
	err := mysqld.InitConfig(cnf)
	if err != nil {
		log.Errorf("%s", err.Error())
		return err
	}
	// Install data dir.
	if err = mysqld.installDataDir(cnf); err != nil {
		return err
	}

	// Start mysqld. We do not use Start, as we have to wait using
	// the root user.
	if err = mysqld.startNoWait(ctx, cnf); err != nil {
		log.Errorf("failed starting mysqld (check mysql error log %v for more info): %v", cnf.ErrorLogPath, err)
		return err
	}

	// Wait for mysqld to be ready, using root credentials, as no
	// user is created yet.
	params := &mysql.ConnParams{
		Uname:      "root",
		Charset:    "utf8",
		UnixSocket: cnf.SocketFile,
	}
	if err = mysqld.wait(ctx, cnf, params); err != nil {
		log.Errorf("failed starting mysqld in time (check mysyql error log %v for more info): %v", cnf.ErrorLogPath, err)
		return err
	}

	// Run initial SQL file.
	sqlFile, err := os.Open(initDBSQLFile)
	if err != nil {
		return fmt.Errorf("can't open init_db_sql_file (%v): %v", initDBSQLFile, err)
	}
	defer sqlFile.Close()
	if err := mysqld.executeMysqlScript(params, sqlFile); err != nil {
		return fmt.Errorf("can't run init_db_sql_file (%v): %v", initDBSQLFile, err)
	}

	return nil
}

// MySQL 5.7 GA and up have deprecated mysql_install_db.
// Instead, initialization is built into mysqld.
func useMysqldInitialize(version string) bool {
	return strings.Contains(version, "Ver 5.7.") ||
		strings.Contains(version, "Ver 8.0.")
}

func (mysqld *Mysqld) installDataDir(cnf *Mycnf) error {
	mysqlRoot, err := vtenv.VtMysqlRoot()
	if err != nil {
		return err
	}
	mysqldPath, err := binaryPath(mysqlRoot, "mysqld")
	if err != nil {
		return err
	}

	mysqlBaseDir, err := vtenv.VtMysqlBaseDir()
	if err != nil {
		return err
	}

	// Check mysqld version.
	_, version, err := execCmd(mysqldPath, []string{"--version"}, nil, mysqlRoot, nil)
	if err != nil {
		return err
	}

	if useMysqldInitialize(version) {
		log.Infof("Installing data dir with mysqld --initialize-insecure")

		args := []string{
			"--defaults-file=" + cnf.path,
			"--basedir=" + mysqlBaseDir,
			"--initialize-insecure", // Use empty 'root'@'localhost' password.
		}
		if _, _, err = execCmd(mysqldPath, args, nil, mysqlRoot, nil); err != nil {
			log.Errorf("mysqld --initialize-insecure failed: %v", err)
			return err
		}
		return nil
	}

	log.Infof("Installing data dir with mysql_install_db")
	args := []string{
		"--defaults-file=" + cnf.path,
		"--basedir=" + mysqlBaseDir,
	}
	cmdPath, err := binaryPath(mysqlRoot, "mysql_install_db")
	if err != nil {
		return err
	}
	if _, _, err = execCmd(cmdPath, args, nil, mysqlRoot, nil); err != nil {
		log.Errorf("mysql_install_db failed: %v", err)
		return err
	}
	return nil
}

func (mysqld *Mysqld) initConfig(root string, cnf *Mycnf, outFile string) error {
	var err error
	var configData string

	env := make(map[string]string)
	envVars := []string{"KEYSPACE", "SHARD", "TABLET_TYPE", "TABLET_ID", "TABLET_DIR", "MYSQL_PORT"}
	for _, v := range envVars {
		env[v] = os.Getenv(v)
	}

	switch hr := hook.NewHookWithEnv("make_mycnf", nil, env).Execute(); hr.ExitStatus {
	case hook.HOOK_DOES_NOT_EXIST:
		log.Infof("make_mycnf hook doesn't exist, reading template files")
		configData, err = cnf.makeMycnf(getMycnfTemplates(root))
	case hook.HOOK_SUCCESS:
		configData, err = cnf.fillMycnfTemplate(hr.Stdout)
	default:
		return fmt.Errorf("make_mycnf hook failed(%v): %v", hr.ExitStatus, hr.Stderr)
	}
	if err != nil {
		return err
	}

	return ioutil.WriteFile(outFile, []byte(configData), 0664)
}

func contains(haystack []string, needle string) bool {
	for _, v := range haystack {
		if v == needle {
			return true
		}
	}
	return false
}

func getMycnfTemplates(root string) []string {
	if *mycnfTemplateFile != "" {
		return []string{*mycnfTemplateFile}
	}

	cnfTemplatePaths := []string{
		path.Join(root, "config/mycnf/default.cnf"),
		path.Join(root, "config/mycnf/master.cnf"),
		path.Join(root, "config/mycnf/replica.cnf"),
	}

	if extraCnf := os.Getenv("EXTRA_MY_CNF"); extraCnf != "" {
		parts := strings.Split(extraCnf, ":")
		cnfTemplatePaths = append(cnfTemplatePaths, parts...)
	}

	switch mysqlFlavor := os.Getenv("MYSQL_FLAVOR"); mysqlFlavor {
	case "MariaDB":
		path := path.Join(root, "config/mycnf/master_mariadb.cnf")
		if !contains(cnfTemplatePaths, path) {
			cnfTemplatePaths = append(cnfTemplatePaths, path)
		}
	case "MariaDB103":
		path := path.Join(root, "config/mycnf/master_mariadb103.cnf")
		if !contains(cnfTemplatePaths, path) {
			cnfTemplatePaths = append(cnfTemplatePaths, path)
		}
	case "MySQL80":
		path := path.Join(root, "config/mycnf/master_mysql80.cnf")
		if !contains(cnfTemplatePaths, path) {
			cnfTemplatePaths = append(cnfTemplatePaths, path)
		}
	default:
		path := path.Join(root, "config/mycnf/master_mysql56.cnf")
		// By default we assume Mysql56 compatable
		if !contains(cnfTemplatePaths, path) {
			cnfTemplatePaths = append(cnfTemplatePaths, path)
		}
	}

	return cnfTemplatePaths
}

// RefreshConfig attempts to recreate the my.cnf from templates, and log and
// swap in to place if it's updated. It keeps a copy of the last version in case fallback is required.
// Should be called from a stable replica, server_id is not regenerated.
func (mysqld *Mysqld) RefreshConfig(ctx context.Context, cnf *Mycnf) error {
	// Execute as remote action on mysqlctld if requested.
	if *socketFile != "" {
		log.Infof("executing Mysqld.RefreshConfig() remotely via mysqlctld server: %v", *socketFile)
		client, err := mysqlctlclient.New("unix", *socketFile)
		if err != nil {
			return fmt.Errorf("can't dial mysqlctld: %v", err)
		}
		defer client.Close()
		return client.RefreshConfig(ctx)
	}

	log.Info("Checking for updates to my.cnf")
	root, err := vtenv.VtRoot()
	if err != nil {
		return err
	}
	f, err := ioutil.TempFile(path.Dir(cnf.path), "my.cnf")
	if err != nil {
		return fmt.Errorf("could not create temp file: %v", err)
	}

	defer os.Remove(f.Name())
	err = mysqld.initConfig(root, cnf, f.Name())
	if err != nil {
		return fmt.Errorf("could not initConfig in %v: %v", f.Name(), err)
	}

	existing, err := ioutil.ReadFile(cnf.path)
	if err != nil {
		return fmt.Errorf("could not read existing file %v: %v", cnf.path, err)
	}
	updated, err := ioutil.ReadFile(f.Name())
	if err != nil {
		return fmt.Errorf("could not read updated file %v: %v", f.Name(), err)
	}

	if bytes.Equal(existing, updated) {
		log.Infof("No changes to my.cnf. Continuing.")
		return nil
	}

	backupPath := cnf.path + ".previous"
	err = os.Rename(cnf.path, backupPath)
	if err != nil {
		return fmt.Errorf("could not back up existing %v: %v", cnf.path, err)
	}
	err = os.Rename(f.Name(), cnf.path)
	if err != nil {
		return fmt.Errorf("could not move %v to %v: %v", f.Name(), cnf.path, err)
	}
	log.Infof("Updated my.cnf. Backup of previous version available in %v", backupPath)

	return nil
}

// ReinitConfig updates the config file as if Mysqld is initializing. At the
// moment it only randomizes ServerID because it's not safe to restore a replica
// from a backup and then give it the same ServerID as before, MySQL can then
// skip transactions in the replication stream with the same server_id.
func (mysqld *Mysqld) ReinitConfig(ctx context.Context, cnf *Mycnf) error {
	log.Infof("Mysqld.ReinitConfig")

	// Execute as remote action on mysqlctld if requested.
	if *socketFile != "" {
		log.Infof("executing Mysqld.ReinitConfig() remotely via mysqlctld server: %v", *socketFile)
		client, err := mysqlctlclient.New("unix", *socketFile)
		if err != nil {
			return fmt.Errorf("can't dial mysqlctld: %v", err)
		}
		defer client.Close()
		return client.ReinitConfig(ctx)
	}

	if err := cnf.RandomizeMysqlServerID(); err != nil {
		return err
	}
	root, err := vtenv.VtRoot()
	if err != nil {
		return err
	}
	return mysqld.initConfig(root, cnf, cnf.path)
}

func (mysqld *Mysqld) createDirs(cnf *Mycnf) error {
	tabletDir := cnf.TabletDir()
	log.Infof("creating directory %s", tabletDir)
	if err := os.MkdirAll(tabletDir, os.ModePerm); err != nil {
		return err
	}
	for _, dir := range TopLevelDirs() {
		if err := mysqld.createTopDir(cnf, dir); err != nil {
			return err
		}
	}
	for _, dir := range cnf.directoryList() {
		log.Infof("creating directory %s", dir)
		if err := os.MkdirAll(dir, os.ModePerm); err != nil {
			return err
		}
		// FIXME(msolomon) validate permissions?
	}
	return nil
}

// createTopDir creates a top level directory under TabletDir.
// However, if a directory of the same name already exists under
// vtenv.VtDataRoot(), it creates a directory named after the tablet
// id under that directory, and then creates a symlink under TabletDir
// that points to the newly created directory.  For example, if
// /vt/data is present, it will create the following structure:
// /vt/data/vt_xxxx /vt/vt_xxxx/data -> /vt/data/vt_xxxx
func (mysqld *Mysqld) createTopDir(cnf *Mycnf, dir string) error {
	tabletDir := cnf.TabletDir()
	vtname := path.Base(tabletDir)
	target := path.Join(vtenv.VtDataRoot(), dir)
	_, err := os.Lstat(target)
	if err != nil {
		if os.IsNotExist(err) {
			topdir := path.Join(tabletDir, dir)
			log.Infof("creating directory %s", topdir)
			return os.MkdirAll(topdir, os.ModePerm)
		}
		return err
	}
	linkto := path.Join(target, vtname)
	source := path.Join(tabletDir, dir)
	log.Infof("creating directory %s", linkto)
	err = os.MkdirAll(linkto, os.ModePerm)
	if err != nil {
		return err
	}
	log.Infof("creating symlink %s -> %s", source, linkto)
	return os.Symlink(linkto, source)
}

// Teardown will shutdown the running daemon, and delete the root directory.
func (mysqld *Mysqld) Teardown(ctx context.Context, cnf *Mycnf, force bool) error {
	log.Infof("mysqlctl.Teardown")
	if err := mysqld.Shutdown(ctx, cnf, true); err != nil {
		log.Warningf("failed mysqld shutdown: %v", err.Error())
		if !force {
			return err
		}
	}
	var removalErr error
	for _, dir := range TopLevelDirs() {
		qdir := path.Join(cnf.TabletDir(), dir)
		if err := deleteTopDir(qdir); err != nil {
			removalErr = err
		}
	}
	return removalErr
}

func deleteTopDir(dir string) (removalErr error) {
	fi, err := os.Lstat(dir)
	if err != nil {
		log.Errorf("error deleting dir %v: %v", dir, err.Error())
		removalErr = err
	} else if fi.Mode()&os.ModeSymlink != 0 {
		target, err := filepath.EvalSymlinks(dir)
		if err != nil {
			log.Errorf("could not resolve symlink %v: %v", dir, err.Error())
			removalErr = err
		}
		log.Infof("remove data dir (symlinked) %v", target)
		if err = os.RemoveAll(target); err != nil {
			log.Errorf("failed removing %v: %v", target, err.Error())
			removalErr = err
		}
	}
	log.Infof("remove data dir %v", dir)
	if err = os.RemoveAll(dir); err != nil {
		log.Errorf("failed removing %v: %v", dir, err.Error())
		removalErr = err
	}
	return
}

// executeMysqlScript executes a .sql script from an io.Reader with the mysql
// command line tool. It uses the connParams as is, not adding credentials.
func (mysqld *Mysqld) executeMysqlScript(connParams *mysql.ConnParams, sql io.Reader) error {
	dir, err := vtenv.VtMysqlRoot()
	if err != nil {
		return err
	}
	name, err := binaryPath(dir, "mysql")
	if err != nil {
		return err
	}
	cnf, err := mysqld.defaultsExtraFile(connParams)
	if err != nil {
		return err
	}
	defer os.Remove(cnf)
	args := []string{
		"--defaults-extra-file=" + cnf,
		"--batch",
	}
	env := []string{
		"LD_LIBRARY_PATH=" + path.Join(dir, "lib/mysql"),
	}
	_, _, err = execCmd(name, args, env, dir, sql)
	if err != nil {
		return err
	}
	return nil
}

// defaultsExtraFile returns the filename for a temporary config file
// that contains the user, password and socket file to connect to
// mysqld.  We write a temporary config file so the password is never
// passed as a command line parameter.  Note ioutil.TempFile uses 0600
// as permissions, so only the local user can read the file.  The
// returned temporary file should be removed after use, typically in a
// 'defer os.Remove()' statement.
func (mysqld *Mysqld) defaultsExtraFile(connParams *mysql.ConnParams) (string, error) {
	var contents string
	if connParams.UnixSocket == "" {
		contents = fmt.Sprintf(`
[client]
user=%v
password=%v
host=%v
port=%v
`, connParams.Uname, connParams.Pass, connParams.Host, connParams.Port)
	} else {
		contents = fmt.Sprintf(`
[client]
user=%v
password=%v
socket=%v
`, connParams.Uname, connParams.Pass, connParams.UnixSocket)
	}

	tmpfile, err := ioutil.TempFile("", "example")
	if err != nil {
		return "", err
	}
	name := tmpfile.Name()
	if _, err := tmpfile.Write([]byte(contents)); err != nil {
		tmpfile.Close()
		os.Remove(name)
		return "", err
	}
	if err := tmpfile.Close(); err != nil {
		os.Remove(name)
		return "", err
	}
	return name, nil
}

// GetAppConnection returns a connection from the app pool.
// Recycle needs to be called on the result.
func (mysqld *Mysqld) GetAppConnection(ctx context.Context) (*dbconnpool.PooledDBConnection, error) {
	return mysqld.appPool.Get(ctx)
}

// GetDbaConnection creates a new DBConnection.
func (mysqld *Mysqld) GetDbaConnection() (*dbconnpool.DBConnection, error) {
	return dbconnpool.NewDBConnection(mysqld.dbcfgs.Dba(), dbaMysqlStats)
}

// GetAllPrivsConnection creates a new DBConnection.
func (mysqld *Mysqld) GetAllPrivsConnection() (*dbconnpool.DBConnection, error) {
	return dbconnpool.NewDBConnection(mysqld.dbcfgs.AllPrivsWithDB(), allprivsMysqlStats)
}

// Close will close this instance of Mysqld. It will wait for all dba
// queries to be finished.
func (mysqld *Mysqld) Close() {
	if mysqld.dbaPool != nil {
		mysqld.dbaPool.Close()
	}
	if mysqld.appPool != nil {
		mysqld.appPool.Close()
	}
}

// OnTerm registers a function to be called if mysqld terminates for any
// reason other than a call to Mysqld.Shutdown(). This only works if mysqld
// was actually started by calling Start() on this Mysqld instance.
func (mysqld *Mysqld) OnTerm(f func()) {
	mysqld.mutex.Lock()
	defer mysqld.mutex.Unlock()
	mysqld.onTermFuncs = append(mysqld.onTermFuncs, f)
}