#!/usr/bin/env python3

# Requirements:
# --- for Debian,Ubuntu series ---
# apt-get install python3 python3-docopt python3-schema
#
# --- for RHEL,CentOS series ---
# yum install     epel-release
# yum install     python3 python3-docopt python3-schema
#
# --- Or, install to ~/.local just for user's environment ---
# (required: python3-pip)
# pip3 install --user docopt schema
#
# --- To use orjson ---
# pip3 install --user orjson

# Coding style check:
# flake8 ./gfptar

import os
import errno
import stat
import sys
import pwd
import grp
import time
import re
import abc
import tarfile
from pprint import pformat
import logging
import subprocess
import concurrent.futures
import threading
from typing import NoReturn
import shutil
from contextlib import contextmanager
import traceback
import unittest
import hashlib
import signal
import resource
import string
import random
from decimal import Decimal, ROUND_DOWN
import sqlite3
import tempfile
import multiprocessing
import queue
import inspect
from collections import defaultdict
import datetime
import atexit
import fnmatch

from docopt import docopt
from schema import Schema, Use, Or


def str2bool(s):
    return s.upper() in ['TRUE', '1', 'ON', 'ENABLE', 'ENABLED']


USE_ORJSON = str2bool(os.getenv('GFPTAR_USE_ORJSON', 'True'))

json_dumps = None
json_loads = None
json_type = None

try:
    if USE_ORJSON:
        import orjson

        json_dumps = orjson.dumps   # byte array
        json_loads = orjson.loads
        json_type = "orjson"
except ImportError:
    pass

if json_type is None:
    import json

    def _dumps_compact(obj):
        return json.dumps(obj, separators=(',', ':')).encode()

    json_dumps = _dumps_compact
    json_loads = json.loads
    json_type = "json"


def format_seconds(seconds, minhour=False):
    if minhour is False:
        return f"{seconds:.0f}s"
    elif seconds < 3600:
        minutes = seconds / 60
        return f"{seconds:.0f}s({minutes:.1f}m)"
    else:
        hours = seconds / 3600
        # Prevent the string from being shortened
        # ex. '1.1h' -> ' 1.1h'
        return f"{seconds:.0f}s({hours:4.1f}h)"


def humanize_number(num, binary_prefix=False):
    if binary_prefix:
        units = ['', 'Ki', 'Mi', 'Gi', 'Ti', 'Pi', 'Ei', 'Zi', 'Yi']
        base = Decimal(1024)
    else:  # SI prefix
        units = ['', 'k', 'M', 'G', 'T', 'P', 'E', 'Z', 'Y']
        base = Decimal(1000)
    if num < base and isinstance(num, int):
        return str(num)
    n = Decimal(num)
    ulen = len(units) - 1
    scale = 0
    while n >= base and scale < ulen:
        n /= base
        scale += 1
    if n < 10:
        # ex. '1.23'
        d = n.quantize(Decimal('0.00'), rounding=ROUND_DOWN)
    elif n < 100:
        # ex. '12.3'
        d = n.quantize(Decimal('0.0'), rounding=ROUND_DOWN)
    elif n < 1000:
        d = n.quantize(Decimal('0'), rounding=ROUND_DOWN)
        # Prevent the string from being shortened
        # ex. ' 123'
        d = f" {d}"
    else:
        # ex. '1234'
        d = n.quantize(Decimal('0'), rounding=ROUND_DOWN)
    return f"{d}{units[scale]}"


def unhumanize_number(numstr, binary_prefix=False):
    if binary_prefix:
        base = 1024
    else:
        base = 1000
    numstrlen = len(numstr)
    if numstrlen == 1:
        return int(numstr)
    lastchar = numstr[-1]
    if lastchar == 'i' and numstrlen > 2:
        # binary prefix: ex. 1Ki
        n = float(numstr[:(numstrlen-2)])
        prefix = numstr[-2]
        base = 1024  # even if binary_prefix=False
    else:  # SI prefix
        n = float(numstr[:(numstrlen-1)])
        prefix = lastchar

    units = {'K': 1,
             'M': 2,
             'G': 3,
             'T': 4,
             'P': 5,
             'E': 6,
             'Z': 7,
             'Y': 8,
             }
    power = units.get(prefix.upper())
    if power is None:
        return int(numstr)
    return int(n * (base ** power))


# Do not use the same DB object from multi-threads.
class DB:
    def __init__(self, filename, check_same_thread=True):
        self.filename = filename
        # NOTE: check_same_thread=True: ignore the following error:
        # SQLite objects created in a thread can only be used
        # in that same thread.
        # The object was created in thread id MM and this is thread id NN.
        self.con = sqlite3.connect(filename,
                                   check_same_thread=check_same_thread)
        self.con.execute('PRAGMA synchronous = OFF')
        self.con.execute('PRAGMA journal_mode = OFF')  # disable rollback
        # self.con.execute('PRAGMA journal_mode = WAL')

    def commit(self):
        self.con.commit()

    def close(self):
        self.con.close()
        self.con = None

    def unlink(self):
        if self.con is not None:
            self.close()
        os.remove(self.filename)


# Interface
class DBObj:
    @classmethod
    def dumps(cls, obj, for_dict):
        raise NotImplementedError

    @classmethod
    def loads(cls, key, txt, for_dict):
        raise NotImplementedError


class JsonObj(DBObj):
    @classmethod
    def dumps(cls, obj, for_dict):
        return json_dumps(obj)

    @classmethod
    def loads(cls, key, txt, for_dict):
        return json_loads(txt)


# Example of DBObj
class IntObj(DBObj):
    def __init__(self, i):
        self.i = i

    @classmethod
    def dumps(cls, obj, for_dict):
        if isinstance(obj, int):
            return str(obj)
        return str(obj.i)

    @classmethod
    def loads(cls, key, txt, for_dict):
        return int(txt)


class StrObj(DBObj):
    def __init__(self, s):
        self.s = s

    @classmethod
    def dumps(cls, obj, for_dict):
        if isinstance(obj, str):
            return obj
        return obj.s

    @classmethod
    def loads(cls, key, txt, for_dict):
        return txt


class FileAttr1(DBObj):
    def __init__(self, mode, mtime, user, group):
        self.mode = mode & 0o7777
        self.mtime = mtime
        self.user = user
        self.group = group

    @classmethod
    def dumps(cls, obj, for_dict):
        array = [obj.mode, obj.mtime, obj.user, obj.group]
        return json_dumps(array)

    @classmethod
    def loads(cls, key, txt, for_dict):
        array = json_loads(txt)
        return cls(array[0], array[1], array[2], array[3])


class FileAttr2(DBObj):
    def __init__(self, mode, mtime, user, group, size, linkname, ftype):
        self.mode = mode & 0o7777
        self.mtime = mtime
        self.user = user
        self.group = group
        self.size = size
        self.linkname = linkname
        self.ftype = ftype

    @classmethod
    def dumps(cls, obj, for_dict):
        array = [obj.mode, obj.mtime, obj.user, obj.group,
                 obj.size, obj.linkname, obj.ftype]
        return json_dumps(array)

    @classmethod
    def loads(cls, key, txt, for_dict):
        array = json_loads(txt)
        return cls(array[0], array[1], array[2], array[3],
                   array[4], array[5], array[6])


class Mtime(DBObj):
    def __init__(self, mtime):
        self.mtime = mtime

    @classmethod
    def dumps(cls, obj, for_dict):
        return str(obj.mtime)

    @classmethod
    def loads(cls, key, txt, for_dict):
        return cls(int(txt))


# Abstract
class DBCollection:
    def __init__(self, db, obj_cls, table_name, clear=False):
        self.db = db
        self.con = db.con
        self.obj_cls = obj_cls
        self.table_name = table_name
        if clear:
            self.clear()
        else:
            self.create_table()

    def obj2txt(self, obj):
        # byte array
        return self.obj_cls.dumps(obj, self.for_dict)

    def txt2obj(self, key, txt):
        if txt is None:
            return None
        return self.obj_cls.loads(key, txt, self.for_dict)

    def create_table(self):
        raise NotImplementedError

    def clear(self):
        self.con.execute(f"DROP TABLE IF EXISTS {self.table_name}")
        self.create_table()

    def commit(self):
        self.db.commit()

    def close(self):
        self.db.close()

    def filename(self):
        return self.db.filename

    def unlink(self):
        self.db.unlink()

    def __len__(self):
        res = self.con.execute(f"SELECT COUNT(*) FROM {self.table_name}")
        return res.fetchone()[0]


class DBDict(DBCollection):
    def create_table(self):
        self.con.execute(f"""
        CREATE TABLE IF NOT EXISTS {self.table_name}
        (key TEXT PRIMARY KEY, value BLOB)
        """)
        self.for_dict = True

    def __setitem__(self, key, value):
        self.con.execute(f"""
        INSERT OR REPLACE INTO {self.table_name} (key, value) VALUES (?, ?)
        """, (key, self.obj2txt(value)))

    def __getitem__(self, key):
        res = self.con.execute(
            f"SELECT value FROM {self.table_name} WHERE key = ?", (key,))
        row = res.fetchone()
        if row:
            return self.txt2obj(key, row[0])
        else:
            raise KeyError(key)

    def __delitem__(self, key):
        if key in self:  # __contains__()
            self.con.execute(
                f"DELETE FROM {self.table_name} WHERE key = ?", (key,))
        else:
            raise KeyError(key)

    def __contains__(self, key):
        res = self.con.execute(
            f"SELECT 1 FROM {self.table_name} WHERE key = ?", (key,))
        row = res.fetchone()
        return row is not None

    def get(self, key, default=None):
        if key in self:  # __contains__()
            return self[key]  # __getitem__()
        return default

    def keys(self):
        res = self.con.execute(f"SELECT key FROM {self.table_name}")
        for row in res:
            yield row[0]

    def values(self):
        res = self.con.execute(f"SELECT key,value FROM {self.table_name}")
        for row in res:
            yield self.txt2obj(row[0], row[1])

    def items(self):
        res = self.con.execute(f"SELECT key,value FROM {self.table_name}")
        for row in res:
            yield row[0], self.txt2obj(row[0], row[1])

    def update(self, other=None, **kwargs):
        if other:
            for key, value in other.items():
                self[key] = value
        for key, value in kwargs.items():
            self[key] = value

    def __iter__(self):
        return self.keys()

    # sort: None, 'ASC', 'DESC'
    def iterator(self, like=None, sort=None, offset=0, limit=-1):
        sql = f"SELECT key,value FROM {self.table_name}"
        if like is not None:
            sql += " WHERE key LIKE ? ESCAPE '\\'"
        if sort is not None:
            if sort.upper() == 'ASC':
                sql += ' ORDER BY key ASC'
            elif sort.upper() == 'DESC':
                sql += ' ORDER BY key DESC'
            else:
                raise AssertionError(f"unknown sort type: {str(sort)}")
        sql += f" LIMIT {limit} OFFSET {offset}"
        if like is not None:
            res = self.con.execute(sql, (like,))
        else:
            res = self.con.execute(sql)
        for row in res:
            yield row[0], self.txt2obj(row[0], row[1])

    @classmethod
    def _escape(cls, txt):
        return txt.replace('%', '\\%').replace('_', '\\_')

    def find_by_prefix(self, prefix, sort=None):
        escaped = self._escape(prefix)
        return self.iterator(like=escaped+'%', sort=sort)


class DBSet(DBCollection):
    def create_table(self):
        self.con.execute(f"""
        CREATE TABLE IF NOT EXISTS {self.table_name}
        (key TEXT PRIMARY KEY)
        """)
        self.for_dict = False

    def update(self, others):
        for key in others:
            self.add(key)

    def add(self, key):
        self.con.execute(f"""
        INSERT OR IGNORE INTO {self.table_name} (key) VALUES (?)
        """, (self.obj2txt(key),))

    def remove(self, key):
        if key in self:
            self.con.execute(f"DELETE FROM {self.table_name} WHERE key = ?",
                             (self.obj2txt(key),))
        else:
            raise KeyError(key)

    def discard(self, key):
        self.con.execute(f"DELETE FROM {self.table_name} WHERE key = ?",
                         (self.obj2txt(key),))

    def __delitem__(self, key):
        self.remove(key)

    def __contains__(self, key):
        res = self.con.execute(f"""
        SELECT 1 FROM {self.table_name} WHERE key = ?
        """, (self.obj2txt(key),))
        row = res.fetchone()
        return row is not None

    def __iter__(self):
        res = self.con.execute(f"SELECT key FROM {self.table_name}")
        for row in res:
            yield self.txt2obj(row[0], row[0])

    # sort: None, 'ASC', 'DESC'
    def iterator(self, sort=None, offset=0, limit=-1):
        sql = f"SELECT key FROM {self.table_name}"
        if sort is not None:
            if sort.upper() == 'ASC':
                sql += ' ORDER BY key ASC'
            elif sort.upper() == 'DESC':
                sql += ' ORDER BY key DESC'
        sql += f" LIMIT {limit} OFFSET {offset}"
        res = self.con.execute(sql)
        for row in res:
            yield self.txt2obj(row[0], row[0])


class DBList(DBCollection):
    ERRMSG_INDEX = 'list index out of range'

    def create_table(self):
        self.con.execute(f"""
        CREATE TABLE IF NOT EXISTS {self.table_name}
        (id INTEGER PRIMARY KEY AUTOINCREMENT, value BLOB)
        """)
        self.for_dict = False

    def __repr__(self):
        return repr(list(self))

    def __str__(self):
        return str(list(self))

    def append(self, obj):
        self.con.execute(f"""
        INSERT INTO {self.table_name} (value) VALUES (?)
        """, (self.obj2txt(obj),))

    def extend(self, lst):
        for obj in lst:
            self.append(obj)

    def __getitem__(self, index):
        res = self.con.execute(f"""
        SELECT id,value FROM {self.table_name} ORDER BY id LIMIT 1 OFFSET ?
        """, (index,))
        row = res.fetchone()
        if row is None:
            raise IndexError(self.ERRMSG_INDEX)
        return self.txt2obj(row[0], row[1])

    def __setitem__(self, index, value):
        with self.con:
            res = self.con.execute(f"""
            SELECT id FROM {self.table_name} ORDER BY id LIMIT 1 OFFSET ?
            """, (index,))
            row = res.fetchone()
            if row is None:
                raise IndexError(self.ERRMSG_INDEX)
            self.con.execute(f"""
            UPDATE {self.table_name} SET value = ? WHERE id = ?
            """, (self.obj2txt(value), row[0]))

    def __delitem__(self, index):
        with self.con:
            res = self.con.execute(f"""
            SELECT id FROM {self.table_name} ORDER BY id LIMIT 1 OFFSET ?
            """, (index,))
            row = res.fetchone()
            if row is None:
                raise IndexError(self.ERRMSG_INDEX)
            self.con.execute(f"""
            DELETE FROM {self.table_name} WHERE id = ?
            """, (row[0],))

    def __iter__(self):
        res = self.con.execute(f"""
        SELECT id,value FROM {self.table_name} ORDER BY id ASC
        """)
        for row in res:
            yield self.txt2obj(row[0], row[1])

    def __reversed__(self):
        res = self.con.execute(f"""
        SELECT id,value FROM {self.table_name} ORDER BY id DESC
        """)
        for row in res:
            yield self.txt2obj(row[0], row[1])

    # sort: None, 'ASC', 'DESC'
    def iterator(self, sort=None, offset=0, limit=-1):
        sql = f"SELECT id,value FROM {self.table_name}"
        if sort is not None:
            if sort.upper() == 'ASC':
                sql += ' ORDER BY id ASC'
            elif sort.upper() == 'DESC':
                sql += ' ORDER BY id DESC'
        sql += f" LIMIT {limit} OFFSET {offset}"
        res = self.con.execute(sql)
        for row in res:
            yield self.txt2obj(row[0], row[1])


class GfException(Exception):
    pass


class GfLogger(logging.getLoggerClass()):
    def __init__(self, name):
        self._mylock = threading.Lock()
        super().__init__(name)

    def _mylog(self, level, msg, *args, exc_info=None):
        frame = inspect.currentframe().f_back.f_back
        lineno = frame.f_lineno
        if exc_info:
            if isinstance(exc_info, BaseException):
                exc_info = (type(exc_info), exc_info, exc_info.__traceback__)
            elif not isinstance(exc_info, tuple):
                exc_info = sys.exc_info()
        record = self.makeRecord(self.name, level, frame.f_code.co_filename,
                                 lineno, msg, args, exc_info)
        with self._mylock:
            self.handle(record)

    def debug(self, msg, *args, **kwargs):
        self._mylog(logging.DEBUG, msg, *args, **kwargs)

    def info(self, msg, *args, **kwargs):
        self._mylog(logging.INFO, msg, *args, **kwargs)

    def warning(self, msg, *args, **kwargs):
        self._mylog(logging.WARNING, msg, *args, **kwargs)

    def error(self, msg, *args, **kwargs):
        self._mylog(logging.ERROR, 'Error: ' + msg, *args, **kwargs)

    def error_exit(self, exit_code, msg, *args, **kwargs):
        self._mylog(logging.ERROR, 'Error: ' + msg, *args, **kwargs)
        sys.exit(exit_code)

    # def fatal(self, msg, *args, **kwargs):
    #     self._mylog(logging.ERROR, 'Error: ' + msg, args)
    #     if 'exit_code' in kwargs:
    #         raise GfException('exit_code={}'.format(
    #             kwargs['exit_code']))
    #     else:
    #         raise GfException('fatal exit')


logger = None


def logger_config(logger, name, loglevel, debug, verbose):
    strm = logging.StreamHandler()  # stderr
    if debug:
        fmt = '%(filename)s:%(levelname)s:L%(lineno)d:' + \
            '%(asctime)s: %(message)s'
        fmt = fmt.format(name)
    elif verbose:
        fmt = '{}:%(levelname)s: %(message)s'.format(name)
    else:
        fmt = '%(message)s'
    formatter_strm = logging.Formatter(fmt=fmt, datefmt='%Y%m%d%H%M%S')
    strm.setFormatter(formatter_strm)
    strm.setLevel(loglevel)
    logger.addHandler(strm)


def logger_init(name, loglevel=logging.WARNING, debug=False, verbose=False):
    global logger

    if logger is not None:
        return logger

    logger = GfLogger('gfptar')
    logger.setLevel(loglevel)
    logger_config(logger, name, loglevel, debug, verbose)
    return logger


def logger_init_without_lock(name, loglevel=logging.WARNING,
                             debug=False, verbose=False):
    logger = logging.RootLogger(loglevel)
    logger_config(logger, name, loglevel, debug, verbose)
    return logger


def get_max_openfiles():
    _, hard_lim = resource.getrlimit(resource.RLIMIT_NOFILE)
    return hard_lim


def update_max_openfiles():
    soft_lim, hard_lim = resource.getrlimit(resource.RLIMIT_NOFILE)
    logger.info(f"current soft RLIMIT_NOFILE: {soft_lim}")
    logger.info(f"current hard RLIMIT_NOFILE: {hard_lim}")
    new_soft_lim = hard_lim
    resource.setrlimit(resource.RLIMIT_NOFILE, (new_soft_lim, hard_lim))
    soft_lim, hard_lim = resource.getrlimit(resource.RLIMIT_NOFILE)
    logger.info(f"new soft RLIMIT_NOFILE: {soft_lim}")


_encoding = sys.getfilesystemencoding()


def get_encoding():
    return _encoding


def set_encoding(enc):
    global _encoding
    if _encoding != enc:
        _encoding = enc


def execcmd_raw(args, stdin=subprocess.DEVNULL, stderr=subprocess.PIPE,
                timeout=None, textmode=False):
    if textmode:
        encoding = get_encoding()
    else:
        encoding = None
    proc = subprocess.Popen(
        args, shell=False, encoding=encoding, close_fds=True,
        stdin=stdin, stdout=subprocess.PIPE, stderr=stderr)
    try:
        out, err = proc.communicate(timeout=timeout)
    except subprocess.TimeoutExpired:
        proc.kill()
        out, err = proc.communicate()
    ret = proc.wait()
    return out, err, ret


use_stderr = True
DEFAULT_STDERR = sys.stderr.buffer


def shutup_stderr():
    global use_stderr
    global DEFAULT_STDERR
    use_stderr = False
    DEFAULT_STDERR = subprocess.DEVNULL
    logger.debug('shutup_stderr')


def execcmd(args, stdin=subprocess.DEVNULL, stderr=subprocess.PIPE,
            timeout=None, textmode=False):
    out, err, ret = execcmd_raw(args, stdin=stdin, stderr=stderr,
                                timeout=timeout, textmode=textmode)
    if ret != 0:
        try:
            err = err.decode().rstrip()
        except MemoryError:
            raise
        except Exception:
            logger.debug('cannot decode: err={}', err)
            pass
        msg = f"{str(args)} (exit={ret}): {err}"
        if err and use_stderr:
            logger.debug(msg)
        raise GfException(msg)
    return out


# text mode
def execcmd_readline(args, stdin=subprocess.DEVNULL):
    if use_stderr:
        stderr = subprocess.PIPE
    else:
        stderr = subprocess.DEVNULL
    proc = subprocess.Popen(
        args, shell=False, encoding=get_encoding(), close_fds=True,
        stdin=stdin, stdout=subprocess.PIPE, stderr=stderr)
    if use_stderr:
        def handle_stderr(process):
            prefix = ''
            if logger.isEnabledFor(logging.INFO):
                prefix = f"({args[0]}) "
            while True:
                eline = process.stderr.readline()
                if eline:
                    eline = eline.rstrip('\r\n')
                    logger.warning(f"{prefix}{eline}")
                elif process.poll() is not None:
                    break

        stderr_thread = threading.Thread(target=handle_stderr, args=(proc,))
        stderr_thread.start()
    else:
        stderr_thread = None
    while True:
        line = proc.stdout.readline()
        if line:
            line = line.rstrip('\r\n')
            yield line
        elif proc.poll() is not None:
            break
    ret = proc.wait()
    if stderr_thread is not None:
        stderr_thread.join()
    if ret != 0:
        raise GfException('{}: returncode={}'.format(' '.join(args), ret))


def gfwhoami():
    return execcmd(['gfwhoami'], textmode=True).strip()


def am_I_gfarmroot():
    out = execcmd(['gfgroup', '-l', 'gfarmroot'], textmode=True)
    patt = r'.* {}( |$)'.format(gfwhoami())
    if re.match(patt, out.strip()):
        return True
    return False


def convert_message(error):
    if hasattr(error, 'filename') and hasattr(error, 'strerror'):
        if error.filename is not None and error.strerror is not None:
            return f"{error.filename}: {error.strerror}"

    def to_str(error):
        if isinstance(error, GfException):  # custom errors
            return f"gfptar error: {str(error)}"
        else:
            return f"{error.__class__.__name__}: {str(error)}"

    message = to_str(error)

    if error.__context__:
        # "During handling of the above exception, another exception occurred"
        message += " (Possible cause: " + to_str(error.__context__) + ")"
    if error.__cause__:
        # "The above exception was the direct cause of the following exception"
        message += " (Direct cause: " + to_str(error.__cause__) + ")"
    return message


class Program(metaclass=abc.ABCMeta):
    def init(self, name) -> NoReturn:
        self.name = name
        self._docopt = docopt(self.getDoc())
        self.opt = self.getSchema().validate(self._docopt)

    def log_init(self):
        self.debug = self.opt['--debug']
        self.verbose = self.opt['--verbose']
        self.quiet = self.opt['--quiet']

        loglevel = logging.WARNING
        if self.debug:
            loglevel = logging.DEBUG
        elif self.verbose:
            loglevel = logging.INFO
        elif self.quiet:
            loglevel = logging.ERROR
        self.loglevel = loglevel

        # use stderr with lock
        logger_init(self.name, loglevel=loglevel, debug=self.debug,
                    verbose=self.verbose)
        logger.debug(pformat(self.opt))

    @abc.abstractmethod
    def getDoc(self) -> str:
        raise NotImplementedError

    @abc.abstractmethod
    def getSchema(self) -> Schema:
        raise NotImplementedError

    @abc.abstractmethod
    def run(self, option) -> NoReturn:
        raise NotImplementedError


class GfURLEntry(DBObj):
    TYPE_FILE = 'F'
    TYPE_DIR = 'D'
    TYPE_SYMLINK = 'S'
    TYPE_OTHER = '?'

    def __init__(self, path, mode, file_type, uname, gname,
                 size, mtime, linkname):
        self.path = path
        self.mode = mode
        self.file_type = file_type
        self.uname = uname
        self.gname = gname
        self.mtime = mtime
        self.linkname = linkname

        if self.is_file():
            self.size = size
        else:
            self.size = 0

    def __str__(self):
        return f"{self.path}, {self.mode:o}, {self.file_type}"

    def __repr__(self):
        return (f"Entry(path={self.path},mode={oct(self.mode)},"
                f"user={self.uname},group={self.gname})")

    # only path must be specified for key when using DBDict
    @classmethod
    def dumps(cls, obj, for_dict):
        # serialize using list() to reduce size
        # [0]...[6]
        array = [obj.mode, obj.file_type, obj.uname, obj.gname,
                 obj.size, obj.mtime, obj.linkname]
        # save path to key when using dict, so don't save path to value
        if not for_dict:
            array.append(obj.path)  # [7]
        return json_dumps(array)

    @classmethod
    def loads(cls, key, txt, for_dict):
        o = json_loads(txt)
        if for_dict:
            path = key
        else:
            path = o[7]
        return cls(path, o[0], o[1], o[2], o[3], o[4], o[5], o[6])

    def url(self):
        return GfURL.init(self.path)

    def subpath(self, baseurl):
        return baseurl.subpath(self.path)

    def size_all(self):
        # include length of path
        this_size = self.size + len(self.path.encode())
        if self.is_symlink():
            this_size += len(self.linkname.encode())
        return this_size

    def is_file(self):
        return self.file_type == self.TYPE_FILE

    def is_directory(self):
        return self.file_type == self.TYPE_DIR

    def is_symlink(self):
        return self.file_type == self.TYPE_SYMLINK

    def toTarinfo(self, path):
        tarinfo = tarfile.TarInfo(path)
        if self.is_file():
            tarinfo.type = tarfile.REGTYPE
        elif self.is_directory():
            tarinfo.type = tarfile.DIRTYPE
        elif self.is_symlink():
            tarinfo.type = tarfile.SYMTYPE
        else:
            logger.warning('unsupported type: %s, %s', path, self.file_type)
            return None
        tarinfo.mode = self.mode & 0o7777
        tarinfo.mtime = self.mtime
        tarinfo.size = self.size
        tarinfo.linkname = self.linkname
        tarinfo.uname = self.uname
        tarinfo.gname = self.gname
        return tarinfo

    def compare(self, ent2, data=False, same_owner=False, bufsize=1048576):
        ent1 = self

        def cmpval(a, b, name):
            if name == 'mtime':
                if USE_GFMKDIR_PLUS is False \
                   or USE_GFCHMOD_PLUS is False \
                   or USE_GFREG_PLUS is False:
                    # skip (not supported)
                    return
                a = int(a)
                b = int(b)
            elif name == 'mode':
                a = oct(a & 0o7777)
                b = oct(b & 0o7777)
            if a != b:
                raise GfException(
                    f"{ent1.path} vs {ent2.path}: prop={name}: {a} != {b}")
            logger.debug('GfURLEntry.compare:prop=%s: PASS', name)

        def cmpprop(ent1, ent2, properties):
            for pname in properties:
                val1 = getattr(ent1, pname)
                val2 = getattr(ent2, pname)
                cmpval(val1, val2, pname)

        # path is not compared
        cmpprop(ent1, ent2, ['file_type', 'mode', 'mtime', 'size', 'linkname'])
        if same_owner:
            cmpprop(ent1, ent2, ['uname', 'gname'])
        if data and ent1.is_file():
            url1 = GfURL.init(ent1.path)
            url2 = GfURL.init(ent2.path)
            if not url1.compare_data(url2, bufsize=bufsize):
                raise GfException(
                    f"{ent1.path} vs {ent2.path}: different data")
            logger.debug('GfURLEntry.compare:data: PASS')
        return True


class GfURL(metaclass=abc.ABCMeta):
    MAXNAMLEN = 255  # SEE ALSO: dirent.h, gfarm/gfs.h (GFS_MAXNAMLEN)
    DEBUG_SLEEP = 0

    @classmethod
    def shutup_stderr(cls):
        shutup_stderr()

    @classmethod
    def init(cls, url, use_gfarm_command=False, local=False, dry_run=False):
        if local:
            return GfURLLocal(url, dry_run)
        if GfURLGfarm.is_my_URL(url):
            return GfURLGfarm(url, dry_run)
        gfurl1 = GfURLLocal(url, dry_run)
        if not use_gfarm_command:
            return gfurl1
        if dry_run:
            # cannot resolve the Gfarm URL from gfarm2fs
            # when creating files.
            # Because the parent directory is not created.
            return gfurl1
        # use_gfarm_command=True: use gf* commands on gfarm2fs
        gfurl2 = gfurl1.get_gfarm_url_by_gfarm2fs()
        if gfurl2 is not None:
            # replace
            logger.debug('use Gfarm URL(%s) on gfarm2fs', gfurl2.url_str)
            return gfurl2
        return gfurl1

    def __init__(self, url, dry_run):
        self._url_str = url
        scheme, host, path = self.parse(url)
        self._scheme = scheme
        self._host = host
        self.path = os.path.normpath(path)
        self.dry_run = dry_run
        if dry_run:
            self.chmod = self.dryrun_chmod
            self.chown = self.dryrun_chown
            self.utime = self.dryrun_utime
            self.mkdir = self.dryrun_mkdir
            self.rename = self.dryrun_rename
            self.remove = self.dryrun_remove
            self.remove_tree = self.dryrun_remove_tree
            self.symlink = self.dryrun_symlink
            self.hardlink = self.dryrun_hardlink
            self.writeopen = self.dryrun_writeopen

    @classmethod
    def parse(cls, url):
        if cls == GfURL:
            supported_classes = [GfURLGfarm]
        else:
            supported_classes = [cls]

        for c in supported_classes:
            if c.SCHEME_LEN > 0 and url.startswith(f"{c.SCHEME}:"):
                # gfarm://host/path -> //host/path
                hostpath = url[(c.SCHEME_LEN+1):]
                scheme = c.SCHEME
                if hostpath.startswith('//'):  # gfarm://...
                    hostpath = hostpath[2:]
                    hp = hostpath.split('/', 1)
                    host = hp[0]
                    if len(hp) >= 2:
                        # ex. gfarm://host/path -> path='/path'
                        # ex. gfarm://host/ -> path='/'
                        path = '/' + hp[1]
                    else:
                        # ex. gfarm://host
                        path = '/'
                    return scheme, host, path
                else:  # gfarm:/... or gfarm:...
                    host = None
                    path = hostpath
                    return scheme, host, path

        # case of GfURLLocal:
        # scheme, host, path
        return None, None, url

    @property
    def basename(self):
        return os.path.basename(self.path)

    @property
    def parent(self):
        parent_path = os.path.dirname(self.path)
        # "." ... relative path (of local path)
        # "/" ... absolute path
        if self.root_url_str != '' \
           and (parent_path == '.' or parent_path == '/'):
            parent_path = ''
        return GfURL.init(self.root_url_str + parent_path,
                          dry_run=self.dry_run)

    @property
    def parent_iter(self):
        def parents_func(u):
            p = u.parent.path
            if p != '' and p != '.' and p != '/':
                yield u.parent
                yield from parents_func(u.parent)
        yield from parents_func(self)

    @property
    def url_str(self):
        return self._url_str

    @property
    def root_url_str(self):
        # ex. gfarm://example.com/a/b/c -> gfarm://example.com
        # ex. gfarm:/abc.def/gh -> gfarm:
        if self._scheme:
            if self._host:
                return f"{self._scheme}://{self._host}"
            else:
                return f"{self._scheme}:"
        return ''

    def subpath(self, fullpath):
        base = self.url_str
        if not fullpath.startswith(base):
            logger.error('subpath: %s, %s', base, fullpath)
            raise AssertionError(f"base={base}, fullpath={fullpath}")
        # logger.debug('subpath: %s, %s', base, fullpath)
        return fullpath[len(base):].lstrip('/')  # relative path

    def url_join(self, subpath):
        return os.path.join(self.url_str, subpath.lstrip('/'))

    def is_gfarm(self):
        return isinstance(self, GfURLGfarm)

    def is_local(self):
        return isinstance(self, GfURLLocal)

    def create_new_dir(self):
        if self.dry_run:
            return
        if self.exists():
            raise FileExistsError(self.url_str)
        self.mkdir()
        if not self.is_directory():
            raise NotADirectoryError(self.url_str)
        if not self.is_empty():
            raise FileExistsError(self.url_str)

    @classmethod
    @abc.abstractmethod
    def is_my_URL(cls, url):
        raise NotImplementedError

    @classmethod
    def set_debug_sleep(cls, sec):
        cls.DEBUG_SLEEP = sec

    def debug_sleep(self):
        if self.DEBUG_SLEEP > 0:
            # time.sleep(0): long time since Python 3.11
            time.sleep(self.DEBUG_SLEEP)

    @abc.abstractmethod
    def chmod(self, mode, mtime=None, user=None, group=None,
              follow_symlinks=True):
        raise NotImplementedError

    def dryrun_chmod(self, mode, mtime=None, user=None, group=None,
                     follow_symlinks=True):
        pass

    @abc.abstractmethod
    def chown(self, user, group, follow_symlinks=True):
        raise NotImplementedError

    def dryrun_chown(self, user, group, follow_symlinks=True):
        pass

    @abc.abstractmethod
    def utime(self, atime, mtime, follow_symlinks=True):
        raise NotImplementedError

    def dryrun_utime(self, atime, mtime, follow_symlinks=True):
        pass

    @abc.abstractmethod
    def mkdir(self, mode=0o700, parents=False):
        raise NotImplementedError

    def dryrun_mkdir(self, mode=0o700, parents=False):
        pass

    def makedirs(self, mode=0o700):
        self.mkdir(mode, parents=True)

    @abc.abstractmethod
    def rename(self, dest):
        raise NotImplementedError

    def dryrun_rename(self, dest):
        pass

    @abc.abstractmethod
    def remove(self):
        raise NotImplementedError

    def dryrun_remove(self):
        pass

    # For test purposes only
    @abc.abstractmethod
    def remove_tree(self, remove_readonly=False):
        raise NotImplementedError

    def dryrun_remove_tree(self, remove_readonly=False):
        pass

    @abc.abstractmethod
    def symlink(self, linkname):
        raise NotImplementedError

    def dryrun_symlink(self, linkname):
        pass

    @abc.abstractmethod
    def hardlink(self, linkname):
        raise NotImplementedError

    def dryrun_hardlink(self, linkname):
        pass

    @abc.abstractmethod
    def exists(self):
        raise NotImplementedError

    @abc.abstractmethod
    def is_directory(self):
        raise NotImplementedError

    @abc.abstractmethod
    def is_symlink(self):
        raise NotImplementedError

    @abc.abstractmethod
    def is_empty(self):
        raise NotImplementedError

    @abc.abstractmethod
    def is_readable(self):
        raise NotImplementedError

    def check_readable(self):
        if not self.is_readable():
            e = PermissionError(13, 'Permission denied', self.url_str)
            logger.debug(str(e))
            raise e

    @abc.abstractmethod
    def is_writable(self):
        raise NotImplementedError

    @abc.abstractmethod
    def size(self):
        raise NotImplementedError

    @abc.abstractmethod
    def mode(self):
        raise NotImplementedError

    @abc.abstractmethod
    def listdir(self, path_only=False, recursive=False, first=False,
                hardlink_warn=True):
        raise NotImplementedError

    @contextmanager
    @abc.abstractmethod
    def readopen(self, textmode=False):
        raise NotImplementedError

    @contextmanager
    @abc.abstractmethod
    def writeopen(self, textmode=False, mode=0o600, mtime=None,
                  user=None, group=None, use_fsync=True, hostname=None):
        raise NotImplementedError

    @contextmanager
    def dryrun_writeopen(self, textmode=False, mode=0o600, mtime=None,
                         user=None, group=None, use_fsync=True, hostname=None):
        self.debug_sleep()
        f = open(os.devnull, 'wb')
        try:
            yield f
        finally:
            f.close()

    def copy_from0(self, inf, bufsize, mode, mtime,
                   user, group, use_fsync, hostname):
        readlen = 0
        with self.writeopen(mode=mode, mtime=mtime,
                            user=user, group=group,
                            use_fsync=use_fsync, hostname=hostname) as outf:
            while True:
                buf = inf.read(bufsize)
                if not buf:
                    break
                # binary mode
                wlen = outf.write(buf)
                readlen += wlen
        return readlen

    def copy_from(self, inf, bufsize, overwrite=False, mode=0o600, mtime=None,
                  user=None, group=None, use_fsync=True, hostname=None):
        try:
            return self.copy_from0(inf, bufsize, mode, mtime,
                                   user, group, use_fsync, hostname)
        except Exception:
            if not overwrite:
                raise
            if not self.exists():
                raise
            if self.is_writable():
                raise
            self.chmod(mode | 0o200)
            return self.copy_from0(inf, bufsize, mode, mtime,
                                   user, group, use_fsync, hostname)

    def sha256(self, bufsize=1048576):
        h = hashlib.sha256()
        with self.readopen() as f:
            while True:
                buf = f.read(bufsize)
                if not buf:
                    break
                h.update(buf)
        return h.hexdigest()

    def compare_data(self, url2, bufsize=1048576):
        url1 = self
        return url1.sha256(bufsize) == url2.sha256(bufsize)

    def compare_entries(self, d2url, data=True, same_owner=False,
                        bufsize=1048576, exclude_list=None):
        def search_compare_pop(subpath, ent1, dirdict2):
            ent2 = dirdict2.pop(subpath, None)
            if ent2 is None:
                return False  # not found
            ent1.compare(ent2, data=data, same_owner=same_owner,
                         bufsize=bufsize)
            # raise if different
            return True

        def excluded(path):
            if exclude_list is None:
                return False
            for exclude in exclude_list:
                if re.match(exclude, path):
                    return True
            return False

        def listdir_ignore_error(gfurl):
            try:
                yield from gfurl.listdir(recursive=True)
            except MemoryError:
                raise
            except Exception as e:
                logger.warning('listdir(%s): error ignored: %s',
                               gfurl.url_str, str(e))

        d1url = self
        # iterator
        d1i = listdir_ignore_error(d1url)
        d2i = listdir_ignore_error(d2url)
        d1dict = dict()
        d2dict = dict()
        d2stop = False
        for d1ent in d1i:
            subpath1 = d1ent.subpath(d1url)
            p1base = os.path.basename(subpath1)
            if excluded(p1base):
                continue
            logger.debug('compare:dir1: %s', subpath1)
            try:
                if not search_compare_pop(subpath1, d1ent, d2dict):
                    # not found, compare later
                    d1dict[subpath1] = d1ent
            except MemoryError:
                raise
            except Exception as e:
                logger.error(str(e))
                return False
            while not d2stop:
                d2ent = next(d2i, None)
                if d2ent is None:
                    d2stop = True
                    break
                subpath2 = d2ent.subpath(d2url)
                p2base = os.path.basename(subpath2)
                if excluded(p2base):
                    continue
                logger.debug('compare:dir2: %s', subpath2)
                try:
                    if not search_compare_pop(subpath2, d2ent, d1dict):
                        # not found, compare later
                        d2dict[subpath2] = d2ent
                except MemoryError:
                    raise
                except Exception as e:
                    logger.error(str(e))
                    return False
        if d2stop is False:
            while True:
                d2ent = next(d2i, None)
                if d2ent is None:
                    break
                subpath2 = d2ent.subpath(d2url)
                p2base = os.path.basename(subpath2)
                if excluded(p2base):
                    continue
                logger.debug('compare:dir2: %s', subpath1)
                d2dict[subpath2] = d2ent
        d1notfound = dict()
        for subpath1, d1ent in d1dict.items():
            try:
                if not search_compare_pop(subpath1, d1ent, d2dict):
                    d1notfound[subpath1] = d1ent
            except MemoryError:
                raise
            except Exception as e:
                logger.error(str(e))
                return False
        if len(d1notfound) > 0 or len(d2dict) > 0:
            logger.error('different entries: dir1=%s, dir2=%s',
                         pformat(d1notfound), pformat(d2dict))
            return False
        logger.debug('compare_entries: PASS')
        return True


USE_GFMKDIR_PLUS = str2bool(os.getenv('GFMKDIR_PLUS', 'True'))
USE_GFCHMOD_PLUS = str2bool(os.getenv('GFCHMOD_PLUS', 'True'))
USE_GFREG_PLUS = str2bool(os.getenv('GFREG_PLUS', 'True'))


class GfURLGfarm(GfURL):
    SCHEME = 'gfarm'
    SCHEME_LEN = 5
    SCHEME_COLON = 'gfarm:'
    # Ex. 12345 -rw-rw-r-- 1 user1  gfarmadm     29 Jan  1 00:00:00 2022 fname
    PAT_ENTRY = re.compile(r'^\s*(\d+)\s+([-dl]\S+)\s+(\d+)\s+(\S+)\s+(\S+)\s+'
                           r'(\d+)\s+(\S+\s+\d+\s+\d+:\d+:\d+\s+\d+)\s+(.+)$')
    PAT_EMPTY = re.compile(r'^\s*$')

    # def __init__(self, url, dry_run):
    #     super().__init__(url, dry_run)

    @classmethod
    def is_my_URL(cls, url):
        return url.startswith(cls.SCHEME_COLON)

    @staticmethod
    def to_oct_str(mode):
        # ex. (int)0o644 -> (str)'644'
        return oct(mode)[2:]

    def chmod(self, mode, mtime=None, user=None, group=None,
              follow_symlinks=True):
        args = ['gfchmod']
        if USE_GFCHMOD_PLUS:
            if mtime is not None:
                args.append('-M')
                args.append(str(int(mtime)))
            if user is not None:
                args.append('-u')
                args.append(user)
            if group is not None:
                args.append('-g')
                args.append(group)
        if not follow_symlinks:
            args.append('-h')
        args.append(self.to_oct_str(mode))
        args.append(self.url_str)
        execcmd(args)
        if not USE_GFCHMOD_PLUS:
            if mtime is not None:
                self.utime(mtime, mtime, follow_symlinks=follow_symlinks)
            self.chown(user, group, follow_symlinks=follow_symlinks)

    def chown(self, user, group, follow_symlinks=True):
        if user is None and group is None:
            return
        chown_args = ['gfchown']
        if not follow_symlinks:
            chown_args.append('-h')
        if user is not None and group is not None:
            chown_args.append(user + ':' + group)
        elif user is not None:
            chown_args.append(user)
        elif group is not None:
            chown_args.append(':' + group)
        chown_args.append(self.url_str)
        execcmd(chown_args)

    def utime(self, atime, mtime, follow_symlinks=True):
        # not used when USE_GFCHMOD_PLUS == True
        logger.warning('mtime cannot be updated (not supported): %s',
                       self.url_str)

    def mkdir(self, mode=0o700, parents=False):
        args = ['gfmkdir']
        if parents:
            args.append('-p')
        if USE_GFMKDIR_PLUS:
            if mode is not None:
                args.append('-m')
                args.append(self.to_oct_str(mode))
            args.append(self.url_str)
            execcmd(args)
        else:
            args.append(self.url_str)
            execcmd(args)
            if mode is not None:
                # slow
                self.chmod(mode)

    def rename(self, dest):
        execcmd(['gfmv', self.url_str, dest])

    def remove(self):
        execcmd(['gfrm', self.url_str])

    def remove_tree(self, remove_readonly=False):
        path = self.url_str
        if path == '/' or path == '':
            raise AssertionError('remove_tree: "/" or ""')
        try:
            execcmd(['gfrm', '-rf', path])
        except MemoryError:
            raise
        except Exception:
            if remove_readonly:
                execcmd(['gfchmod', '-R', '700', path])
                execcmd(['gfrm', '-rf', path])
            else:
                raise

    def symlink(self, linkname):
        execcmd(['gfln', '-s', linkname, self.url_str])

    def hardlink(self, linkname):
        execcmd(['gfln', linkname, self.url_str])

    def exists(self):
        out, err, ret = execcmd_raw(['gftest', '-e', self.url_str])
        return ret == 0

    def is_directory(self):
        out, err, ret = execcmd_raw(['gftest', '-d', self.url_str])
        return ret == 0

    def is_symlink(self):
        out, err, ret = execcmd_raw(['gftest', '-h', self.url_str])
        return ret == 0

    def is_empty(self):
        count = 0
        for line in execcmd_readline(['gfls', '-1a', self.url_str]):
            if line is None:
                continue
            if line == '.' or line == '..':
                continue
            count += 1
        logger.debug('is_empty: %s, %d', self.url_str, count)
        return count == 0

    R_OK = 0o4
    W_OK = 0o2
    X_OK = 0o1

    def access(self, mode):
        out, err, ret = execcmd_raw(['gfxattr', '-g', self.url_str,
                                     'gfarm.effective_perm'])
        if ret == 0:
            val = int.from_bytes(out, byteorder='big')
            return val & mode > 0
        raise GfException(f"{self.url_str}: gfarm.effective_perm: {err}")

    def is_readable(self):
        return self.access(self.R_OK)

    def is_writable(self):
        return self.access(self.W_OK)

    def size(self):
        for entry in self.listdir(recursive=False, myself_only=True):
            logger.debug('size: %s, %d', entry.path, entry.size)
            return entry.size

    def mode(self):
        for entry in self.listdir(recursive=False, myself_only=True):
            logger.debug('mode: %s, %o', entry.path, entry.mode)
            return entry.mode & 0o777

    @classmethod
    def from_rwx(cls, rwx, highchar):
        perm = 0
        highbit = 0
        r = rwx[0]
        w = rwx[1]
        x = rwx[2]
        if r == 'r':
            perm |= 0o4
        if w == 'w':
            perm |= 0o2
        if x == 'x':
            perm |= 0o1
        elif x == highchar:
            perm |= 0o1
            highbit = 0o1
        elif x == highchar.upper():
            highbit = 0o1
        return perm, highbit

    @classmethod
    def mode_convert(cls, mode_str, name):
        typestr = mode_str[0]
        file_type = None
        if typestr == 'd':
            file_type = GfURLEntry.TYPE_DIR
        elif typestr == 'l':
            file_type = GfURLEntry.TYPE_SYMLINK
        elif typestr == '-':
            file_type = GfURLEntry.TYPE_FILE
        else:
            logger.warning('unsupported type: %s, %s', name, typestr)

        mode = 0
        perm, highbit = cls.from_rwx(mode_str[1:4], 's')
        mode |= (perm << 6)
        mode |= (highbit << 11)
        perm, highbit = cls.from_rwx(mode_str[4:7], 's')
        mode |= (perm << 3)
        mode |= (highbit << 10)
        perm, highbit = cls.from_rwx(mode_str[7:10], 't')
        mode |= perm
        mode |= (highbit << 9)
        return mode, file_type

    def listdir(self, path_only=False, recursive=False, first=False,
                hardlink_warn=False, myself_only=False):
        dirname = self.url_str
        gfls_opt = '-ailT'
        if recursive:
            gfls_opt += 'R'
        if myself_only:
            gfls_opt += 'd'
        for line in execcmd_readline(['gfls', gfls_opt, self.url_str]):
            logger.debug('listdir: raw line=%s', line)
            if line is None:
                continue
            if self.PAT_EMPTY.match(line):
                continue
            m = self.PAT_ENTRY.match(line)
            if m:
                # Ex.
                # 12345 -rw-rw-r-- 1 user1 group1 29 Jan 1 00:00:00 2022 fname
                inum = int(m.group(1))
                mode_str = m.group(2)
                nlink = int(m.group(3))
                uname = m.group(4)
                gname = m.group(5)
                size = int(m.group(6))
                mtime_str = m.group(7)
                name = m.group(8)

                if name == '..':
                    continue
                if not first and name == '.':
                    continue
                mtime = time.mktime(
                    time.strptime(mtime_str, '%b %d %H:%M:%S %Y'))
                mode, file_type = self.mode_convert(mode_str, name)
                if file_type is None:
                    continue
                linkname = ''
                if file_type == GfURLEntry.TYPE_SYMLINK:
                    link = name.split(' -> ')
                    name = link[0]
                    linkname = link[1]
                elif file_type == GfURLEntry.TYPE_FILE:
                    if hardlink_warn and nlink >= 2:
                        logger.warning('hard link is not supported: '
                                       'nlink=%d, inode=%d (Gfarm): %s',
                                       nlink, inum, name)
                if first and name == '.':
                    path = dirname
                elif '/' in name:  # "name" is fullpath of a file
                    path = name
                else:
                    path = dirname + '/' + name
                # path is Gfarm URL (gfarm:/...)
                yield GfURLEntry(path, mode, file_type, uname, gname,
                                 size, mtime, linkname)
            else:
                # ex. gfarm:/home/user1/dir: -> gfarm:/home/user1/dir
                dirname = line[:-1]
                if not dirname.startswith(self.url_str):
                    raise AssertionError(f"{dirname}.startswith("
                                         f"{self.url_str})")
                first = False

    def gfsched(self, is_file=False, write_mode=False, number=None):
        args = ['gfsched']
        if is_file:
            args.append('-f')
            args.append(self.url_str)
        else:
            args.append('-P')
            args.append(self.url_str)
        if write_mode is not None:
            args.append('-w')
        if number is not None:
            args.append('-n')
            args.append(str(number))
        return [line for line in execcmd_readline(args)]

    def gfreg(self, textmode=False, mode=None, mtime=None,
              user=None, group=None, hostname=None):
        class Gfreg():
            def __init__(self, url, proc, stdin):
                self.url = url
                self.proc = proc
                self.stdin = stdin

            def close(self):
                self.stdin.close()

            def post(self):
                if USE_GFREG_PLUS:
                    return
                # must be called after wait()
                if mode is not None:
                    self.url.chmod(mode, mtime=mtime, user=user, group=group)
                else:
                    if mtime is not None:
                        self.url.utime(mtime, mtime)
                    self.url.chown(user, group)

        args = ['gfreg']
        if USE_GFREG_PLUS:
            if mode is not None:
                args.append('-m')
                args.append(self.to_oct_str(mode))
            if mtime is not None:
                args.append('-M')
                args.append(str(int(mtime)))
            if user is not None:
                args.append('-u')
                args.append(user)
            if group is not None:
                args.append('-g')
                args.append(group)
        if hostname:
            args.append('-h')
            args.append(hostname)
        args.append('-')  # stdin
        args.append(self.url_str)
        if textmode:
            encoding = get_encoding()
        else:
            encoding = None
        proc = subprocess.Popen(
            args, shell=False, encoding=encoding, close_fds=True,
            stdin=subprocess.PIPE, stdout=subprocess.DEVNULL,
            stderr=DEFAULT_STDERR)
        return Gfreg(self, proc, proc.stdin)

    def gfexport(self, textmode=False):
        args = ['gfexport', self.url_str]
        if textmode:
            encoding = get_encoding()
        else:
            encoding = None
        return subprocess.Popen(
            args, shell=False, encoding=encoding, close_fds=True,
            stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
            stderr=DEFAULT_STDERR)

    @contextmanager
    def readopen(self, textmode=False):
        self.debug_sleep()
        proc = self.gfexport(textmode=textmode)
        try:
            yield proc.stdout
        finally:
            proc.stdout.close()
            ret = proc.wait()
            if ret != 0:
                raise GfException('{}: returncode={}'.format(
                    ' '.join(proc.args), ret))

    @contextmanager
    def writeopen(self, textmode=False, mode=0o600, mtime=None,
                  user=None, group=None, use_fsync=True, hostname=None):
        self.debug_sleep()
        # TODO XXX add gfreg option for use_fsync
        gfreg_obj = self.gfreg(textmode=textmode, mode=mode, mtime=mtime,
                               user=user, group=group, hostname=hostname)
        proc = gfreg_obj.proc
        try:
            yield gfreg_obj.stdin
        finally:
            gfreg_obj.close()
            ret = proc.wait()
            if ret != 0:
                raise GfException('{}: returncode={}'.format(
                    ' '.join(proc.args), ret))
            gfreg_obj.post()


def get_uid(user):
    try:
        pw = pwd.getpwnam(user)
        if pw is not None:
            return pw.pw_uid
    except KeyError:
        pass
    return -1  # uid is not changed by os.chown()


def get_gid(group):
    try:
        gr = grp.getgrnam(group)
        if gr is not None:
            return gr.gr_gid
    except KeyError:
        pass
    return -1  # gid is not changed by os.chown()


class GfURLLocal(GfURL):
    SCHEME = ''
    SCHEME_LEN = 0

    @classmethod
    def is_my_URL(cls, url):
        return True

    def get_gfarm_url_by_gfarm2fs(self):
        # SEE ALSO: lib/libgfarm/gfarm/gfarm_path.c
        XATTR_GFARM2FS_URL = 'gfarm2fs.url'
        try:
            val = os.getxattr(self.url_str, XATTR_GFARM2FS_URL)
            gfurl = GfURL.init(val.decode(get_encoding()),
                               dry_run=self.dry_run)
            if gfurl.is_gfarm():
                return gfurl
            raise GfException('unexpected Gfarm URL format: ' + gfurl.url_str)
        except OSError as e:
            if e.errno == errno.EOPNOTSUPP:
                return None
            elif e.errno == errno.ENOENT:
                parent_path = self.parent.url_str
                if parent_path == '':
                    # local filesystem
                    return None
                # creation mode (parent must exist)
                try:
                    val = os.getxattr(parent_path, XATTR_GFARM2FS_URL)
                    parent_str = val.decode(get_encoding())
                    gfurl = GfURL.init(os.path.join(parent_str, self.basename),
                                       dry_run=self.dry_run)
                    if gfurl.is_gfarm():
                        return gfurl
                    raise GfException('unexpected Gfarm URL format: '
                                      + gfurl.url_str)
                except OSError as e2:
                    if e2.errno == errno.EOPNOTSUPP:
                        # local filesystem
                        return None
                    raise
                except Exception:
                    raise
            else:
                raise
        except Exception:
            raise

    def chmod(self, mode, mtime=None, user=None, group=None,
              follow_symlinks=True):
        if follow_symlinks:
            os.chmod(self.url_str, mode)
        else:
            try:
                os.chmod(self.url_str, mode, follow_symlinks=False)
            except NotImplementedError:
                if not self.is_symlink():
                    raise
                # ignore, not changed for symlink
        if mtime is not None:
            self.utime(mtime, mtime, follow_symlinks=follow_symlinks)
        self.chown(user, group)

    def chown(self, user, group, follow_symlinks=False):
        if user is not None or group is not None:
            # shutil.chown does not have follow_symlinks option.
            uid = get_uid(user)
            gid = get_gid(group)
            os.chown(self.url_str, uid, gid, follow_symlinks=follow_symlinks)

    def utime(self, atime, mtime, follow_symlinks=True):
        os.utime(self.url_str, times=(atime, mtime),
                 follow_symlinks=follow_symlinks)

    def mkdir(self, mode=0o700, parents=False):
        if parents:
            try:
                os.mkdir(self.url_str, mode)
                return  # OK
            except FileExistsError:
                return  # OK
            except Exception:
                pass
            # Reference: https://docs.python.org/3/library/os.html#os.makedirs
            # Changed in version 3.7: The mode argument no longer
            # affects the file permission bits of newly created
            # intermediate-level directories.
            # os.makedirs(self.url_str, mode=mode, exist_ok=True)
            for p in reversed(list(self.parent_iter)):
                if p.exists():
                    continue
                try:
                    os.mkdir(p.url_str, mode | 0o700)
                except FileExistsError:  # race condition
                    pass  # ignored
                except Exception:
                    raise
            os.mkdir(self.url_str, mode)
        else:
            os.mkdir(self.url_str, mode)

    def _chmod_recursive(self, path, dir_mode, file_mode):
        for root, dirs, files in os.walk(path):
            for d in dirs:
                dir_path = os.path.join(root, d)
                os.chmod(dir_path, dir_mode)
            for f in files:
                file_path = os.path.join(root, f)
                os.chmod(file_path, file_mode)

    def rename(self, dest):
        os.rename(self.path, dest)

    def remove(self):
        os.remove(self.path)

    def remove_tree(self, remove_readonly=False):
        path = self.path
        if path == '/' or path == '':
            raise AssertionError('remove_tree: "/" or ""')
        if remove_readonly:
            self._chmod_recursive(path, 0o700, 0o600)
        shutil.rmtree(path)

    def symlink(self, linkname):
        os.symlink(linkname, self.url_str)

    def hardlink(self, linkname):
        os.link(linkname, self.url_str)

    def exists(self):
        return os.path.exists(self.url_str)

    def is_directory(self):
        return os.path.isdir(self.url_str)

    def is_symlink(self):
        return os.path.islink(self.url_str)

    def is_empty(self):
        return len(os.listdir(self.url_str)) == 0

    def is_readable(self):
        return os.access(self.url_str, os.R_OK)

    def is_writable(self):
        return os.access(self.url_str, os.W_OK)

    def size(self):
        st = os.stat(self.url_str, follow_symlinks=False)
        logger.debug('size: %s, %d', self.url_str, st.st_size)
        return st.st_size

    def mode(self):
        st = os.stat(self.url_str, follow_symlinks=False)
        logger.debug('mode: %s, %o', self.url_str, st.st_mode)
        return st.st_mode & 0o777

    @classmethod
    def _readlink(cls, path, is_symlink):
        if is_symlink:
            linkname = os.readlink(path)
        else:
            linkname = ''
        return linkname

    @classmethod
    def uid2name(cls, uid):
        try:
            pw = pwd.getpwuid(uid)
            return pw.pw_name
        except Exception:
            return str(uid)

    @classmethod
    def gid2name(cls, gid):
        try:
            gr = grp.getgrgid(gid)
            return gr.gr_name
        except Exception:
            return str(gid)

    @classmethod
    def _toFileType(cls, st):
        file_type = GfURLEntry.TYPE_OTHER
        if stat.S_ISREG(st.st_mode):
            file_type = GfURLEntry.TYPE_FILE
        elif stat.S_ISDIR(st.st_mode):
            file_type = GfURLEntry.TYPE_DIR
        elif stat.S_ISLNK(st.st_mode):
            file_type = GfURLEntry.TYPE_SYMLINK
        return file_type

    def _toGfURLEntry(self, entry, path_only, hardlink_warn):
        if path_only:
            return GfURLEntry(entry.path, 0, None,
                              'root', 'root',
                              0, 0, '')
        st = entry.stat(follow_symlinks=False)
        file_type = self._toFileType(st)
        if file_type == GfURLEntry.TYPE_FILE:
            if hardlink_warn and st.st_nlink >= 2:
                logger.warning('hard link is not supported: '
                               'nlink=%d, inode=%d (Local): %s',
                               st.st_nlink, st.st_ino, entry.path)
        linkname = self._readlink(entry.path, entry.is_symlink())
        return GfURLEntry(entry.path, st.st_mode, file_type,
                          self.uid2name(st.st_uid),
                          self.gid2name(st.st_gid),
                          st.st_size, st.st_mtime, linkname)

    def _scandir(self, path, path_only, recursive, first, hardlink_warn):
        if first:
            st = os.stat(path, follow_symlinks=True)
            yield GfURLEntry(path, st.st_mode, self._toFileType(st),
                             self.uid2name(st.st_uid),
                             self.gid2name(st.st_gid),
                             st.st_size, st.st_mtime, '')
            file_type = self._toFileType(st)
            if file_type != GfURLEntry.TYPE_DIR:
                return

        first = False
        has_error = None
        try:
            for entry in os.scandir(path):
                yield self._toGfURLEntry(entry, path_only, hardlink_warn)
                if recursive and entry.is_dir(follow_symlinks=False):
                    try:
                        yield from self._scandir(entry.path, path_only,
                                                 recursive,
                                                 first, hardlink_warn)
                    except Exception as e2:
                        if has_error is None:  # save first error
                            has_error = e2
                        # DO NOT raise here
        except Exception as e:
            if has_error is None:  # save first error
                has_error = e
                logger.warning(convert_message(e))  # logging only once
        if has_error is not None:
            raise has_error

    def listdir(self, path_only=False, recursive=False, first=False,
                hardlink_warn=True):
        yield from self._scandir(self.url_str, path_only, recursive, first,
                                 hardlink_warn)

    @contextmanager
    def readopen(self, textmode=False):
        self.debug_sleep()
        if textmode:
            f = open(self.url_str, 'rt', encoding=get_encoding())
        else:
            f = open(self.url_str, 'rb')
        try:
            yield f
        finally:
            f.close()

    @contextmanager
    def writeopen(self, textmode=False, mode=0o600, mtime=None,
                  user=None, group=None, use_fsync=True, hostname=None):
        self.debug_sleep()
        tmpmode = mode | 0o200  # necessary (Permission denied at ex.0o400)
        fd = os.open(path=self.url_str,
                     flags=(os.O_WRONLY | os.O_CREAT | os.O_TRUNC),
                     mode=tmpmode,
                     )
        if textmode:
            f = open(fd, 'wt', encoding=get_encoding(), closefd=True)
        else:
            f = open(fd, 'wb', closefd=True)
        try:
            yield f
        finally:
            if use_fsync:
                f.flush()
                os.fsync(f.fileno())
            f.close()
            self.chmod(mode, mtime=mtime, user=user, group=group)


class Compress:
    TYPE_NO = 'no'
    TYPE_GZIP = 'gz'
    TYPE_BZIP2 = 'bz2'
    TYPE_XZ = 'xz'
    compress_type = None
    prog_for_create = None
    prog_for_extract = None

    @classmethod
    def init(cls, compress_type, compress_prog,
             gzip_prog, bzip2_prog, xz_prog):
        if gzip_prog is not None:
            cls.gzip_prog = shutil.which(gzip_prog)
        else:
            cls.gzip_prog = None
        if bzip2_prog is not None:
            cls.bzip2_prog = shutil.which(bzip2_prog)
        else:
            cls.bzip2_prog = None
        if xz_prog is not None:
            cls.xz_prog = shutil.which(xz_prog)
        else:
            cls.xz_prog = None

        cls.compress_type = compress_type
        if compress_prog is not None:
            compress_prog0 = shutil.which(compress_prog)
            if compress_prog0 is None:
                raise GfptarError(f"{compress_prog}: command not found.")
            compress_prog = compress_prog0
            cls.prog_for_create = compress_prog
            cls.prog_for_extract = compress_prog
        else:
            # auto selection
            if compress_type == cls.TYPE_GZIP:
                compress_prog = cls.gzip_prog
            elif compress_type == cls.TYPE_BZIP2:
                compress_prog = cls.bzip2_prog
            elif compress_type == cls.TYPE_XZ:
                compress_prog = cls.xz_prog
            else:
                compress_prog = None
            cls.prog_for_create = compress_prog
            cls.prog_for_extract = None  # use get_compress_prog()

        logger.debug('compress_type=%s', cls.compress_type)
        logger.debug('prog_for_create=%s', cls.prog_for_create)
        logger.debug('prog_for_extract=%s', cls.prog_for_extract)
        logger.debug('gzip_prog=%s', cls.gzip_prog)
        logger.debug('bzip2_prog=%s', cls.bzip2_prog)
        logger.debug('xz_prog=%s', cls.xz_prog)

    @classmethod
    def get_compress_prog(cls, name):
        if name.endswith('.gz'):
            return cls.gzip_prog
        elif name.endswith('.bz2'):
            return cls.bzip2_prog
        elif name.endswith('.xz'):
            return cls.xz_prog
        else:
            return None

    @classmethod
    def compress(cls, compress_prog, outf):
        args = [compress_prog]
        # binary mode
        return subprocess.Popen(
            args, shell=False, close_fds=True,
            stdin=subprocess.PIPE, stdout=outf,
            stderr=DEFAULT_STDERR)

    @classmethod
    def decompress(cls, compress_prog, inf):
        args = [compress_prog, '-d']
        # binary mode
        return subprocess.Popen(
            args, shell=False, close_fds=True,
            stdin=inf, stdout=subprocess.PIPE,
            stderr=DEFAULT_STDERR)


class GfTarFile(tarfile.TarFile):
    ATTR_PROC_LIST = '_gfptar_proc_list'  # [(proc, fileobj, fileobj), ...]
    ATTR_USE_FSYNC = 'use_fsync'
    METHOD_add_entry = 'add_entry'

    @classmethod
    def extract_open(cls, gfurl, copybufsize, compress_prog=None):
        if compress_prog is None:
            compress_prog = Compress.prog_for_extract
        if compress_prog is None:
            compress_prog = Compress.get_compress_prog(gfurl.url_str)
        logger.debug('extract_open: compress_prog=%s', compress_prog)
        # use Stream (not seekable)
        if compress_prog is None \
           and Compress.compress_type != Compress.TYPE_NO:
            openmode = 'r|*'  # any type (gz, bz2, xz)
        else:
            openmode = 'r|'
        if not gfurl.exists():
            raise FileNotFoundError(gfurl.url_str)
        # list of tuple(proc, closeable obj, synchronizable obj)
        proc_list = []
        gfurl.check_readable()
        if gfurl.is_gfarm():
            if compress_prog:
                gfexport_proc = gfurl.gfexport()
                decompress_proc = Compress.decompress(compress_prog,
                                                      gfexport_proc.stdout)
                tar = cls.open(None, mode=openmode,
                               fileobj=decompress_proc.stdout,
                               copybufsize=copybufsize)
                proc_list.append(tuple([gfexport_proc,
                                        gfexport_proc.stdout, None]))
                proc_list.append(tuple([decompress_proc,
                                        decompress_proc.stdout, None]))
            else:
                gfexport_proc = gfurl.gfexport()
                tar = cls.open(None, mode=openmode,
                               fileobj=gfexport_proc.stdout,
                               copybufsize=copybufsize)
                proc_list.append(tuple([gfexport_proc,
                                        gfexport_proc.stdout, None]))
        else:
            if compress_prog:
                inf = open(gfurl.url_str, 'rb')
                decompress_proc = Compress.decompress(compress_prog, inf)
                tar = cls.open(None, mode=openmode,
                               fileobj=decompress_proc.stdout,
                               copybufsize=copybufsize)
                proc_list.append(tuple([None, inf, None]))
                proc_list.append(tuple([decompress_proc,
                                        decompress_proc.stdout, None]))
            else:
                tar = GfTarFile.open(gfurl.url_str, mode=openmode,
                                     copybufsize=copybufsize)
        setattr(tar, cls.ATTR_PROC_LIST, proc_list)
        return tar

    @classmethod
    def create_open(cls, gfurl, copybufsize, compress_prog=None,
                    use_fsync=True, target_host=None,
                    dummy_input=False):
        if compress_prog is None:
            compress_prog = Compress.prog_for_create
        # use Stream (not seekable)
        if compress_prog is None \
           and Compress.compress_type != Compress.TYPE_NO:
            openmode = 'w|' + Compress.compress_type
        else:
            openmode = 'w|'
        if gfurl.exists():
            raise FileExistsError(gfurl.url_str)
        # list of tuple(proc, closeable obj, synchronizable obj)
        proc_list = []
        if gfurl.is_gfarm() and not gfurl.dry_run:
            if compress_prog:
                gfreg_obj = gfurl.gfreg(mode=0o600, hostname=target_host)
                compress_proc = Compress.compress(compress_prog,
                                                  gfreg_obj.stdin)
                tar = cls.open(None, mode=openmode,
                               fileobj=compress_proc.stdin,
                               copybufsize=copybufsize)
                proc_list.append(tuple([compress_proc, compress_proc.stdin,
                                        None]))
                proc_list.append(tuple([gfreg_obj.proc, gfreg_obj, None]))
            else:
                gfreg_obj = gfurl.gfreg(mode=0o600, hostname=target_host)
                tar = cls.open(None, mode=openmode,
                               fileobj=gfreg_obj.stdin,
                               copybufsize=copybufsize)
                proc_list.append(tuple([gfreg_obj.proc, gfreg_obj, None]))
        else:  # Local or dry_run
            if gfurl.dry_run:
                outf = open(os.devnull, 'wb')
            else:
                outf = open(gfurl.url_str, 'wb')
            if compress_prog:
                compress_proc = Compress.compress(compress_prog, outf)
                tar = cls.open(None, mode=openmode,
                               fileobj=compress_proc.stdin,
                               copybufsize=copybufsize)
                proc_list.append(tuple([compress_proc, compress_proc.stdin,
                                        None]))
            else:
                tar = cls.open(gfurl.url_str, mode=openmode,
                               fileobj=outf, copybufsize=copybufsize)
            if gfurl.dry_run:
                proc_list.append((None, outf, None))
            else:
                proc_list.append((None, outf, outf))
            gfurl.chmod(0o600)
        logger.debug('GfTarFile.create_open: %s', gfurl.url_str)
        setattr(tar, cls.ATTR_PROC_LIST, proc_list)
        setattr(tar, cls.ATTR_USE_FSYNC, use_fsync)
        if dummy_input:
            setattr(tar, cls.METHOD_add_entry, tar._add_entry_dummy)
        else:
            setattr(tar, cls.METHOD_add_entry, tar._add_entry)
        return tar

    # override
    def close(self):
        super().close()
        use_fsync = getattr(self, self.ATTR_USE_FSYNC, True)
        proc_list = getattr(self, self.ATTR_PROC_LIST, None)
        if proc_list:
            for proc_tuple in proc_list:
                proc, close_obj, sync_obj = proc_tuple
                if use_fsync and sync_obj:
                    sync_obj.flush()
                    os.fsync(sync_obj.fileno())
            for proc_tuple in proc_list:
                proc, close_obj, sync_obj = proc_tuple
                if close_obj:
                    close_obj.close()
            for proc_tuple in proc_list:
                proc, close_obj, sync_obj = proc_tuple
                if proc is not None:
                    logger.debug('close external process for tar: %s',
                                 str(proc.args))
                    ret = proc.wait()
                    if ret != 0:
                        raise GfException('{}: returncode={}'.format(
                            ' '.join(proc.args), ret))
            for proc_tuple in proc_list:
                proc, close_obj, sync_obj = proc_tuple
                if close_obj:
                    post_func = getattr(close_obj, 'post', None)
                    if post_func:
                        close_obj.post()

    def _add_entry_dummy(self, subpath, entry):
        tarinfo = entry.toTarinfo(subpath)
        if tarinfo is None:  # warning, skip
            return
        if entry.is_file():
            with RandomStream(entry.size) as f:
                self.addfile(tarinfo, fileobj=f)
        else:
            self.addfile(tarinfo)
        return tarinfo

    # NOTE: add() is not expected behavior. Use addfile() instead.
    # - add() can copy a hard link,
    #   but a hard link cannot be extracted from gfexport (stream open)
    # - When the specified MEMBER for --create is a symlink,
    #   the entry will be archived as symlink.
    def _add_entry(self, subpath, entry):
        tarinfo = entry.toTarinfo(subpath)
        if tarinfo is None:  # warning, skip
            return
        # hard link is not supported
        if entry.is_file():
            url = GfURL.init(entry.path)
            url.check_readable()
            with url.readopen() as f:
                self.addfile(tarinfo, fileobj=f)
        else:
            self.addfile(tarinfo)
        return tarinfo


class RandomStream:
    def __init__(self, size):
        self.size = size
        self.written = 0
        self.rand_block_size = 1024
        self.rand_data = os.urandom(self.rand_block_size)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        pass

    def read(self, size=-1):
        if self.written >= self.size:
            return b''
        if size == -1:
            size = self.size
        remaining_size = self.size - self.written
        read_size = min(remaining_size, size)
        repeat_count = read_size / self.rand_block_size
        repeat_remainder = read_size % self.rand_block_size
        data = self.rand_data * int(repeat_count) \
            + self.rand_data[:repeat_remainder]
        self.written += read_size
        return data


class TestGfptar(unittest.TestCase):
    @staticmethod
    def suite():
        suite = unittest.TestSuite()
        suite.addTest(
            unittest.defaultTestLoader.loadTestsFromTestCase(TestGfptar))
        return suite

    def test_humanize(self):
        self.assertEqual(humanize_number(0.001), '0.00')
        self.assertEqual(humanize_number(0.1), '0.10')
        self.assertEqual(humanize_number(10.11), '10.1')
        self.assertEqual(humanize_number(100.11), ' 100')
        self.assertEqual(humanize_number(1023, binary_prefix=True), '1023')
        self.assertEqual(humanize_number(1024, binary_prefix=True), '1.00Ki')
        self.assertEqual(humanize_number(1), '1')
        self.assertEqual(humanize_number(999), '999')
        self.assertEqual(humanize_number(1000), '1.00k')
        self.assertEqual(humanize_number(99999), '99.9k')
        self.assertEqual(humanize_number(100000), ' 100k')
        self.assertEqual(humanize_number(999999), ' 999k')
        self.assertEqual(humanize_number(1000000), '1.00M')
        self.assertEqual(humanize_number(6100000000000000000), '6.10E')

    def test_unhumanize(self):
        self.assertEqual(unhumanize_number('999'), 999)
        self.assertEqual(unhumanize_number('1k'), 1000)
        self.assertEqual(unhumanize_number('1K'), 1000)
        self.assertEqual(unhumanize_number('1.234k'), 1234)
        self.assertEqual(unhumanize_number('1k', binary_prefix=True), 1024)
        self.assertEqual(unhumanize_number('1ki'), 1024)
        self.assertEqual(unhumanize_number('1Ki'), 1024)
        self.assertEqual(unhumanize_number('2Mi'), 2097152)
        self.assertEqual(unhumanize_number('3Gi'), 3221225472)
        self.assertEqual(unhumanize_number('4Ti'), 4398046511104)
        self.assertEqual(unhumanize_number('5Pi'), 5629499534213120)
        self.assertEqual(unhumanize_number('6Ei'), 6917529027641081856)
        self.assertEqual(unhumanize_number('6.1E'), 6100000000000000000)

    def test_GfURL_use_gfarm_command_for_local(self):
        url = GfURL.init('/tmp', use_gfarm_command=True)
        self.assertEqual(url.is_local(), True)
        self.assertEqual(url.path, '/tmp')

    def test_GfURL_parse(self):
        self.assertEqual(GfURL.parse('gfarm://host'),
                         ('gfarm', 'host', '/'))
        self.assertEqual(GfURL.parse('gfarm://host/'),
                         ('gfarm', 'host', '/'))
        self.assertEqual(GfURL.parse('gfarm://host/path'),
                         ('gfarm', 'host', '/path'))
        self.assertEqual(GfURL.parse('gfarm://host/path://'),
                         ('gfarm', 'host', '/path://'))
        self.assertEqual(GfURL.parse('gfarm://host/path;abc'),
                         ('gfarm', 'host', '/path;abc'))
        self.assertEqual(GfURL.parse('gfarm://host/path#abc'),
                         ('gfarm', 'host', '/path#abc'))
        self.assertEqual(GfURL.parse('gfarm://host/path?abc'),
                         ('gfarm', 'host', '/path?abc'))
        self.assertEqual(GfURL.parse('gfarm:/path/abc'),
                         ('gfarm', None, '/path/abc'))
        self.assertEqual(GfURL.parse('gfarm:path/abc'),
                         ('gfarm', None, 'path/abc'))
        self.assertEqual(GfURL.parse('/abc/#def:/'),
                         (None, None, '/abc/#def:/'))
        self.assertEqual(GfURL.parse('abc:/def://'),
                         (None, None, 'abc:/def://'))
        self.assertEqual(GfURLLocal.parse('gfarm:abc:/def://'),
                         (None, None, 'gfarm:abc:/def://'))

    def test_root_url_str(self):
        self.assertEqual(GfURL.init('gfarm://host/abc').root_url_str,
                         "gfarm://host")
        self.assertEqual(GfURL.init('gfarm://host:port/abc').root_url_str,
                         "gfarm://host:port")
        self.assertEqual(GfURL.init('gfarm:///abc').root_url_str,
                         "gfarm:")
        self.assertEqual(GfURL.init('gfarm:/abc').root_url_str,
                         "gfarm:")
        self.assertEqual(GfURL.init('/abc').root_url_str,
                         "")

    def test_url_parent(self):
        self.assertEqual(
            GfURL.init('gfarm://host/abc').parent.url_str,
            "gfarm://host")
        self.assertEqual(GfURL.init('gfarm://host/abc/def').parent.url_str,
                         "gfarm://host/abc")
        self.assertEqual(GfURL.init('gfarm://host:port/abc').parent.url_str,
                         "gfarm://host:port")
        self.assertEqual(
            GfURL.init('gfarm://host:port/abc/def').parent.url_str,
            "gfarm://host:port/abc")
        self.assertEqual(GfURL.init('gfarm:///abc').parent.url_str,
                         "gfarm:")
        self.assertEqual(GfURL.init('gfarm:///abc/def').parent.url_str,
                         "gfarm:/abc")
        self.assertEqual(GfURL.init('gfarm:/abc').parent.url_str,
                         "gfarm:")
        self.assertEqual(GfURL.init('gfarm:/abc/def').parent.url_str,
                         "gfarm:/abc")
        self.assertEqual(GfURL.init('/abc').parent.url_str,
                         "/")
        self.assertEqual(GfURL.init('/abc/def').parent.url_str,
                         "/abc")
        self.assertEqual(GfURL.init('abc/def').parent.url_str,
                         "abc")
        self.assertEqual(GfURL.init('./abc/def').parent.url_str,
                         "abc")

    def test_DBDict(self):
        euid = os.geteuid()
        tmpdir = tempfile.TemporaryDirectory(prefix=f"gfptar-test-{euid}-",
                                             dir=None)
        testdb = DB(os.path.join(tmpdir.name, 'test.db'))
        d = DBDict(testdb, IntObj, 'test_dict')

        d.update({'a%%bc': IntObj(1), 'a__bc': IntObj(2), 'bc': 3})
        self.assertEqual(d['a%%bc'], 1)
        self.assertEqual(d['a__bc'], 2)
        self.assertEqual(d['bc'], 3)
        self.assertEqual(d.get('bc'), 3)
        self.assertEqual(d.get('d'), None)
        self.assertNotIn('d', d)
        self.assertIn('a%%bc', d)

        self.assertEqual(set(k for k, v in d.find_by_prefix('a%%b')),
                         {'a%%bc'})
        self.assertEqual(set(k for k, v in d.find_by_prefix('a__b')),
                         {'a__bc'})
        # sort order: % _
        self.assertEqual([k for k, v in d.find_by_prefix('a', sort='ASC')],
                         ['a%%bc', 'a__bc'])
        self.assertEqual([k for k, v in d.find_by_prefix('a', sort='DESC')],
                         ['a__bc', 'a%%bc'])

        d.clear()
        d['a'] = 4
        self.assertEqual(d['a'], 4)

        d.clear()
        d.update({'a': IntObj(1)})
        d['a'] = IntObj(2)
        self.assertEqual(d['a'], 2)

        d.clear()
        d.update({'a': IntObj(1)})
        del d['a']
        self.assertNotIn('a', d)

        d.unlink()

    def test_DBSet(self):
        euid = os.geteuid()
        tmpdir = tempfile.TemporaryDirectory(prefix=f"gfptar-test-{euid}-",
                                             dir=None)
        testdb = DB(os.path.join(tmpdir.name, 'test.db'))
        s = DBSet(testdb, IntObj, 'test_set')

        s.update({IntObj(1), IntObj(2), IntObj(3)})
        self.assertEqual(set(s), {1, 2, 3})

        s.clear()
        s.add(IntObj(1))
        self.assertIn(IntObj(1), s)
        self.assertIn(1, s)

        s.clear()
        s.update({IntObj(1), IntObj(2), IntObj(3)})
        s.add(IntObj(4))
        self.assertIn(IntObj(4), s)

        s.clear()
        s.update({IntObj(1), IntObj(2), IntObj(3)})
        s.remove(IntObj(2))
        self.assertNotIn(IntObj(2), s)

        with self.assertRaises(KeyError):
            s.remove(IntObj(4))
        s.discard(IntObj(4))

        s.unlink()

    def test_DBList(self):
        euid = os.geteuid()
        tmpdir = tempfile.TemporaryDirectory(prefix=f"gfptar-test-{euid}",
                                             dir=None)
        testdb = DB(os.path.join(tmpdir.name, 'test.db'))
        lst = DBList(testdb, IntObj, 'test_list')
        lst.extend([IntObj(1), IntObj(2), IntObj(3)])
        self.assertEqual(list(lst), [1, 2, 3])

        lst.clear()
        lst.append(IntObj(1))
        self.assertEqual(lst[0], 1)

        lst.clear()
        lst.extend([IntObj(1), IntObj(2), IntObj(3)])
        lst[1] = IntObj(4)
        self.assertEqual(list(lst), [1, 4, 3])

        lst.clear()
        lst.extend([IntObj(1), IntObj(2), IntObj(3)])
        del lst[1]
        self.assertEqual(list(lst), [1, 3])

        lst.unlink()


class GfptarError(GfException):
    pass


class CannotBeArchivedError(GfptarError):
    def __init__(self, files=0, directories=0):
        msg = ''
        if files >= 1:
            msg += f"{files} file"
            if files >= 2:
                msg += 's'
        if directories >= 1:
            if msg:
                msg += ' and '
            msg += f"{directories} "
            if directories >= 2:
                msg += 'directories'
            else:
                msg += 'directory'
            msg += '(readdir)'
        self.message = f"{msg} cannot be archived"
        super().__init__(self.message)


@contextmanager
def ignore_exception(ignore):
    try:
        yield
    except Exception as e:
        if ignore:
            logger.debug('Ignored', exc_info=e)
            return
        raise


class GfptarProgram(Program):
    LIST_SUFFIX = '.lst'  # to ignore old files
    TMP_SUFFIX = '.tmp'
    SERIAL_FORMAT = '%04d'

    def __init__(self, name):
        self.init(name)
        self.canceled = False
        self.lock_init(False)
        self.futures = None
        self.hardlink_warn = True
        self.tmpdir = None
        self.test_mode = False
        self.worker_list = []  # (started, process, inq, outq)
        self.progress_enabled = True

    def options_init(self):
        self.log_init()
        if self.quiet:
            GfURL.shutup_stderr()
        logger.debug('USE_GFMKDIR_PLUS: %s', USE_GFMKDIR_PLUS)
        logger.debug('USE_GFCHMOD_PLUS: %s', USE_GFCHMOD_PLUS)
        logger.debug('USE_GFREG_PLUS: %s', USE_GFREG_PLUS)

        logger.debug(f'JSON library: {json_type}')

        set_encoding(self.opt['--encoding'])
        self.jobs = self.opt['--jobs']
        # SEE ALSO: MT_enabled()
        if self.jobs >= 1:
            self.max_workers = self.jobs
        else:
            self.max_workers = 1
        self.bufsize = self.opt['--bufsize']
        self.use_fsync = not self.opt['--disable-fsync']
        self.gen_db = self.opt['--generate-db']
        self.enable_tempdb_aggregation \
            = not self.opt['--disable-tempdb-aggregation']
        self.max_tempdb_aggregation \
            = self.opt['--max-entries-for-tempdb-aggregation']
        self.workdir = self.opt['--workdir']
        euid = os.geteuid()
        # Temporary files are removed when the process exits.
        # dir=None: system default
        if self.tmpdir:
            self.tmpdir.cleanup()
        self.tmpdir = tempfile.TemporaryDirectory(prefix=f"gfptar-{euid}-",
                                                  dir=self.workdir)
        self.sig_init()  # to clean tmpdir

        self.progress_enabled = self._progress_enabled()
        self.progress_interval = self.opt['--progress-interval']
        progress_unit_type = self.opt['--progress-unit']
        if progress_unit_type == 'si':
            self._humanize = self._humanize_si
        elif progress_unit_type == 'bin':
            self._humanize = self._humanize_bin
        elif progress_unit_type == 'raw':
            self._humanize = self._humanize_raw
        else:
            self._humanize = self._humanize_si

        self.memory_limit = self.opt['--memory']
        if self.memory_limit is not None:
            self.set_memory_limit(self.memory_limit)

        self.compress_type_specified = self.opt['--type']
        if self.compress_type_specified is None:
            self.compress_type = Compress.TYPE_GZIP
        else:
            self.compress_type = self.compress_type_specified

        self.compress_prog = self.opt['--use-compress-program']
        self.gzip_prog = self.opt['--gzip-program']
        self.xz_prog = self.opt['--xz-program']
        self.bzip2_prog = self.opt['--bzip2-program']
        Compress.init(self.compress_type, self.compress_prog,
                      self.gzip_prog, self.bzip2_prog, self.xz_prog)
        if Compress.gzip_prog is None:
            raise GfException('gzip: command not found. Please install gzip.')

        self.dry_run = self.opt['--dry-run']

        self.debug_sleep = self.opt['--debug-sleep']
        GfURL.set_debug_sleep(self.debug_sleep)

        self.exclude_re_list = []
        self.exclude = self.opt['--exclude']
        if isinstance(self.exclude, str):
            reobj = re.compile(fnmatch.translate(self.exclude))
            self.exclude_re_list.append(reobj)
        elif isinstance(self.exclude, list):
            for ex in self.exclude:
                reobj = re.compile(fnmatch.translate(ex))
                self.exclude_re_list.append(reobj)

        self.exclude_from = self.opt['--exclude-from']
        if self.exclude_from is not None:
            logger.debug(f"exclude_from={self.exclude_from}")
            exclude_url = GfURL.init(self.exclude_from)
            with exclude_url.readopen(textmode=True) as f:
                while True:
                    line = f.readline().strip()
                    if not line:
                        break
                    if len(line) == 0:
                        continue
                    logger.debug(f"exclude_from: line=`{line}`")
                    reobj = re.compile(fnmatch.translate(line))
                    self.exclude_re_list.append(reobj)

        if len(self.exclude_re_list) > 0:
            self.exclude_match = self.exclude_match0
        else:
            self.exclude_match = self.exclude_match_noop

    def exclude_match0(self, path):
        if path.startswith('./'):  # equal to removeprefix('./')
            path = path[2:]
        # logger.debug(f"exclude_match0(): path={path}")
        if any(reobj.match(path) for reobj in self.exclude_re_list):
            logger.debug(f"exclude_match0()=True: path={path}")
            return True
        return False

    def exclude_match_noop(self, path):
        return False

    def set_memory_limit(self, max_memory):
        resource.setrlimit(resource.RLIMIT_AS, (max_memory, max_memory))

    def sig_init(self, message=True):
        def sig_handler(signum, frame):
            if message:
                sys.stderr.write(f"\nInterrupt (signal={signum})\n")
            self.canceled = True

        signal.signal(signal.SIGHUP, sig_handler)
        signal.signal(signal.SIGINT, sig_handler)
        signal.signal(signal.SIGTERM, sig_handler)
        signal.signal(signal.SIGPIPE, signal.SIG_IGN)

    def sig_default(self):
        signal.signal(signal.SIGHUP, signal.SIG_DFL)
        signal.signal(signal.SIGINT, signal.SIG_DFL)
        signal.signal(signal.SIGTERM, signal.SIG_DFL)
        signal.signal(signal.SIGPIPE, signal.SIG_IGN)

    def sig_ignore(self):
        signal.signal(signal.SIGHUP, signal.SIG_IGN)
        signal.signal(signal.SIGINT, signal.SIG_IGN)
        signal.signal(signal.SIGTERM, signal.SIG_IGN)
        signal.signal(signal.SIGPIPE, signal.SIG_IGN)

    def getDoc(self) -> str:
        return __doc__

    def getSchema(self) -> Schema:
        return _schema

    def _progress_enabled(self):
        return not self.quiet

    def run(self):
        try:
            outdir = self.opt['--create']
            if outdir:
                basedir = self.opt['--basedir']
                infiles = self.opt['MEMBER']
                self.cmd_create(outdir, basedir, infiles)
                return

            outdir = self.opt['--append']
            if outdir:
                basedir = self.opt['--basedir']
                infiles = self.opt['MEMBER']
                self.cmd_append(outdir, basedir, infiles)

            outdir = self.opt['--update']
            if outdir:
                basedir = self.opt['--basedir']
                infiles = self.opt['MEMBER']
                self.cmd_update(outdir, basedir, infiles)
                return

            outdir = self.opt['--extract']
            if outdir:
                indir = self.opt['INDIR']
                members = self.opt['MEMBER']
                self.cmd_extract(outdir, indir, members)
                return

            indir = self.opt['--list']
            if indir:
                if self.opt['--verbose']:
                    self.cmd_list_verbose(indir)
                else:
                    self.cmd_list_simple(indir)
                return

            if self.opt['--test']:
                self.test_main_short()
                return
            if self.opt['--test-long']:
                self.test_main_long()
                return
        except Exception as e:
            self.cancel()
            # NOTE: If threads are canceled, child processes may not stop.
            #       In that case, gfptar may freeze.
            # if self.futures is not None:
            #     for t in self.futures:
            #         logger.debug('Canceled: %s', self.futures[t])
            #         t.cancel()
            if self.debug:
                raise
            else:
                if issubclass(type(e), CannotBeArchivedError):
                    # already reported
                    pass
                else:
                    sys.stderr.write('\n')
                    logger.error(convert_message(e))
                sys.exit(1)
        finally:
            self.worker_terminate()
            self.sig_default()
            if self.tmpdir:
                self.tmpdir.cleanup()
                logger.debug('tmpdir.cleanup')
            logger.debug('DONE')

    def test_init(self):
        self.options_init()
        self.am_I_gfarmroot = am_I_gfarmroot()
        if self.am_I_gfarmroot:
            logger.error('warning: gfarmroot is enabled: '
                         + '"noread" files are not used')
        self.hardlink_warn = False
        self.uid = os.getuid()
        self.pid = os.getpid()
        out = gfwhoami()
        self.gfarm_user = out.strip()
        self.test_mode = True
        self.test_workdir_init()

    def test_main_short(self):
        self.test_init()
        # self.test_unit()
        self.test_opt_pattern(full=False)
        self.test_member()
        self.test_update_append_exclude()
        self.test_specified_dir()

    def test_main_long(self):
        self.test_init()
        self.test_unit()
        self.test_opt_pattern(full=True)
        self.test_member()
        self.test_update_append_exclude()
        # self.test_invalid('url', 'gfarm:/tmp', 'dst', False)  # allowed
        self.test_invalid('dot1', '.', 'dst', True)
        self.test_invalid('dot2', '', 'dst', False)
        self.test_invalid('dot3', './', 'dst', False)
        self.test_invalid('dot4', '././././', 'dst', False)
        self.test_invalid('dotdot1', '..', 'dst', True)
        self.test_invalid('dotdot2', '../', 'dst', False)
        self.test_invalid('dotdot3', '../abc', 'dst', False)
        self.test_invalid('dotdot4', './..', 'dst', False)

    def test_unit(self):
        verbosity = 2
        runner = unittest.TextTestRunner(verbosity=verbosity)
        result = runner.run(TestGfptar.suite())
        if len(result.errors) > 0 or len(result.failures) > 0:
            logger.error_exit(1, 'unittest error')
        print('unittest ... PASS')

    def test_opt_pattern(self, full=False):
        save_opt_size = self.opt['--size']
        save_opt_jobs = self.opt['--jobs']
        save_opt_type = self.opt['--type']
        save_opt_compress_prog = self.opt['--use-compress-program']
        save_opt_gendb = self.opt['--generate-db']
        save_opt_verbose = self.opt['--verbose']

        self.opt['--use-compress-program'] = None

        # create tar per one entry
        self.opt['--size'] = 0

        # test --generate-db
        # --verbose: test report_path (SEE ALSO: create_a_tar_process1)
        self.opt['--verbose'] = True
        self.opt['--generate-db'] = True
        self.test_simple('generate_db')
        self.opt['--generate-db'] = save_opt_gendb
        self.opt['--verbose'] = save_opt_verbose

        # test --jobs
        if full:
            # jobs == 0: not use multi-thread
            pattern_jobs = [0, 16]
        else:
            pattern_jobs = []  # skip
        for jobs in pattern_jobs:
            self.opt['--jobs'] = jobs
            self.test_simple('jobs_' + str(jobs), use_all_files=True)
        self.opt['--jobs'] = save_opt_jobs

        # create one tar
        self.opt['--size'] = unhumanize_number('100M')

        # test --type
        if full:
            pattern_type = [
                'gz',
                'bz2',
                'xz',
                'no']
        else:
            pattern_type = ['gz']
        for t in pattern_type:
            self.opt['--type'] = t
            self.test_simple('type_' + t)
        self.opt['--type'] = save_opt_type

        # test --type and --use-compress-program
        if full:
            pattern_compress_prog = {
                'gzip': 'gz',
                'pigz': 'gz',
                'bzip2': 'bz2',
                'xz': 'xz',
                'pbzip2': 'bz2',
                'lzip': 'lz',
                'lzop': 'lzo',
                'lz4': 'lz4',
            }
        else:
            pattern_compress_prog = {
                'xz': 'xz',
            }
        for prog, t in pattern_compress_prog.items():
            w = shutil.which(prog)
            if not w:
                logger.error('SKIPPED: No such command: %s', prog)
                continue
            self.opt['--type'] = t
            self.opt['--use-compress-program'] = prog
            self.test_simple('compress_prog_' + prog)

        self.opt['--size'] = save_opt_size
        self.opt['--jobs'] = save_opt_jobs
        self.opt['--type'] = save_opt_type
        self.opt['--use-compress-program'] = save_opt_compress_prog

    def test_workdir_init(self):
        def remove_tree(url):
            logger.debug('remove_tree: %s', url.url_str)
            url.remove_tree(remove_readonly=True)

        test_dname = f"gfptar-test-{self.gfarm_user}-{self.uid}-{self.pid}"
        self.testdir_gfarm = os.path.join(self.opt['--test-workdir-gfarm'],
                                          test_dname)
        self.testdir_gfarm_url = GfURL.init(self.testdir_gfarm)
        self.testdir_gfarm_url.mkdir()
        atexit.register(remove_tree, self.testdir_gfarm_url)

        self.testdir_local = os.path.join(self.opt['--test-workdir-local'],
                                          test_dname)
        self.testdir_local_url = GfURL.init(self.testdir_local)
        self.testdir_local_url.mkdir()
        atexit.register(remove_tree, self.testdir_local_url)

    def test_workdir_set(self, dname, gfarm=True, local=True):
        if gfarm:
            self.workdir_gfarm = os.path.join(self.testdir_gfarm, dname)
            self.workdir_gfarm_url = GfURL.init(self.workdir_gfarm)
            self.workdir_gfarm_url.mkdir()
        if local:
            self.workdir_local = os.path.join(self.testdir_local, dname)
            self.workdir_local_url = GfURL.init(self.workdir_local)
            self.workdir_local_url.mkdir()

    def test_invalid(self, name, src, dst, for_gfarm):
        testname = f"invalid-{name}"
        self.test_workdir_set(testname, gfarm=for_gfarm, local=not for_gfarm)
        if for_gfarm:
            workdir = self.workdir_gfarm
            workdir_url = self.workdir_gfarm_url
        else:
            workdir = self.workdir_local
            workdir_url = self.workdir_local_url

        testsrc_name = src
        # srcdir = workdir_url.url_join(testsrc_name)
        # srcdir_url = GfURL.init(srcdir)
        # srcdir_url.mkdir()

        test1_name = dst
        test1_dir = workdir_url.url_join(test1_name)

        ok = False
        try:
            self.cmd_create(test1_dir, workdir, [testsrc_name])
        except GfException as e:
            self.print_trace(e)
            if str(e).startswith('specifying '):
                ok = True
            else:
                raise

        workdir_url.remove_tree(remove_readonly=True)
        if ok:
            print(testname + ' ... PASS')
        else:
            print(testname + ' ... FAIL (unexpected success)')

    def test_simple(self, name, use_all_files=False):
        testname = f"simple-{name}"
        self.test_workdir_set(testname)
        workdir_local = self.workdir_local
        workdir_gfarm = self.workdir_gfarm
        workdir_local_url = self.workdir_local_url
        workdir_gfarm_url = self.workdir_gfarm_url

        testsrc_name = 'test-src'
        srcdir_local = workdir_local_url.url_join(testsrc_name)
        srcdir_gfarm = workdir_gfarm_url.url_join(testsrc_name)
        if use_all_files:
            readonly = True
            if self.am_I_gfarmroot:
                noread = False
            else:
                noread = True
            link = True
            longname = True
        else:
            readonly = False
            noread = False
            link = False
            longname = False
        count1 = self.test_prepare_srcdir(srcdir_local,
                                          readonly, noread, link, longname)
        count2 = self.test_prepare_srcdir(srcdir_gfarm,
                                          readonly, noread, link, longname)
        if count1 != count2:
            logger.error(f"count1({count1}) != count2({count2})")
            logger.error_exit(1, testname + '(count1 != count2) ' +
                              '... FAIL (test_prepare_srcdir is broken)')
        test1_name = 'test-1-create'
        test1_dir_gfarm = workdir_gfarm_url.url_join(test1_name)
        test2_name = 'test-2-extract'
        test2_dir_gfarm = workdir_gfarm_url.url_join(test2_name)
        test3_name = 'test-3-create'
        test3_dir_local = workdir_local_url.url_join(test3_name)
        test4_name = 'test-4-extract'
        test4_dir_local = workdir_local_url.url_join(test4_name)

        # Gfarm -> Gfarm(tar)
        try:
            # from srcdir_gfarm
            self.cmd_create(test1_dir_gfarm, workdir_gfarm, [testsrc_name])
            if noread:
                logger.error_exit(1, testname + '(create:Gfarm->Gfarm) ' +
                                  '... FAIL (unexpected success)')
        except GfptarError as e:
            if not noread:
                raise
            logger.info(f"IGNORED: {str(e)}")
        # Gfarm(tar) -> Gfarm
        self.cmd_extract(test2_dir_gfarm, test1_dir_gfarm, [])
        # Gfarm -> Local(tar)
        try:
            self.cmd_create(test3_dir_local, test2_dir_gfarm, [testsrc_name])
            if noread:
                logger.error_exit(1, testname + '(create:Gfarm->Local) ' +
                                  '... FAIL (unexpected success)')
        except GfptarError as e:
            if not noread:
                raise
            logger.info(f"IGNORED: {str(e)}")
        # Local(tar) -> Local
        self.cmd_extract(test4_dir_local, test3_dir_local, [])

        # --list for gfarm
        count = self.cmd_list_simple(test1_dir_gfarm)
        if count != count1:
            logger.error(f"count({count}) != count1({count1})")
            logger.error_exit(1, testname +
                              '(list_simple(test1_dir_gfarm)) ... FAIL')
        # --verbose --list for local
        count = self.cmd_list_verbose(test3_dir_local)
        if count != count1:
            logger.error(f"count({count}) != count1({count1})")
            logger.error_exit(1, testname +
                              '(list_verbose(test3_dir_local)) ... FAIL')

        if readonly:
            # extract a member (SEE ALSO: test_prepare_srcdir)
            member = testsrc_name + '/dir1/readonly/file#2'
            all_members = [testsrc_name,
                           testsrc_name + '/dir1',
                           testsrc_name + '/dir1/readonly',
                           member,
                           ]
            test_member_g_name = 'test-gfptar-member-g'
            test_member_l_name = 'test-gfptar-member-l'
            test_member_g = workdir_gfarm_url.url_join(test_member_g_name)
            test_member_l = workdir_local_url.url_join(test_member_l_name)
            self.cmd_extract(test_member_g, test1_dir_gfarm, [member])
            self.cmd_extract(test_member_l, test3_dir_local, [member])
            g_member = GfURL.init(os.path.join(test_member_g, member))
            l_member = GfURL.init(os.path.join(test_member_l, member))
            if not g_member.compare_data(l_member):
                logger.error_exit(1, testname
                                  + '(extract a member) ' +
                                  '... FAIL (data mismatch)')
            for testdir in [test_member_g, test_member_l]:
                url = GfURL.init(testdir)
                for ent in url.listdir(recursive=True):
                    p = ent.subpath(url)
                    if p in all_members:
                        continue
                    logger.error('unexpected member: %s', p)
                    logger.error_exit(1, testname + '(extract a member) ' +
                                      '... FAIL (unexpected member found)')

        test5_name = 'test-5-create'
        test5_dir_gfarm = workdir_gfarm_url.url_join(test5_name)
        test6_name = 'test-6-extract'
        test6_dir_local = workdir_local_url.url_join(test6_name)

        # NOTE: check ignoring hardlinks for Local
        # Local -> Gfarm(tar)
        try:
            self.cmd_create(test5_dir_gfarm, workdir_local, [testsrc_name])
            if noread:
                logger.error_exit(1, testname + '(create:Local->Gfarm) ' +
                                  '... FAIL (unexpected success)')
        except GfptarError as e:
            if not noread:
                raise
            logger.info(f"IGNORED: {str(e)}")
        # Gfarm(tar) -> Local
        self.cmd_extract(test6_dir_local, test5_dir_gfarm, [])

        test4_srcdir_local = os.path.join(test4_dir_local, testsrc_name)
        if not self.test_compare_local(test4_srcdir_local, srcdir_local,
                                       same_owner=True):
            logger.error_exit(1, testname + ' ... FAIL (different data)(1)')

        if self.test_compare_local(test4_dir_local, test6_dir_local,
                                   same_owner=True):
            print(testname + ' ... PASS')
            workdir_local_url.remove_tree(remove_readonly=True)
            workdir_gfarm_url.remove_tree(remove_readonly=True)
        else:
            logger.error_exit(1, testname + ' ... FAIL (different data)(2)')

    def test_member(self):
        testname = 'gfptar-test-member'
        self.test_workdir_set(testname)
        # workdir_local = self.workdir_local
        # workdir_gfarm = self.workdir_gfarm
        workdir_local_url = self.workdir_local_url
        workdir_gfarm_url = self.workdir_gfarm_url

        testsrc_name = 'test-src'
        srcdir_local = workdir_local_url.url_join(testsrc_name)
        srcdir_gfarm = workdir_gfarm_url.url_join(testsrc_name)
        self.test_prepare_srcdir(srcdir_local, readonly=True)
        self.test_prepare_srcdir(srcdir_gfarm, readonly=True)

        test1_name = 'test-1-create'
        test1_dir_gfarm = workdir_gfarm_url.url_join(test1_name)
        test2_name = 'test-2-extract'
        test2_dir_gfarm = workdir_gfarm_url.url_join(test2_name)
        test3_name = 'test-3-create'
        test3_dir_local = workdir_local_url.url_join(test3_name)
        test4_name = 'test-4-extract'
        test4_dir_local = workdir_local_url.url_join(test4_name)
        test5_name = 'test-5-extract'
        test5_dir_local = workdir_local_url.url_join(test5_name)
        test6_name = 'test-6-extract'
        test6_dir_local = workdir_local_url.url_join(test6_name)

        # pick files as members (SEE ALSO: test_prepare_srcdir)
        # 'gfarm:abc' is a filename (not Gfarm URL)
        members = ['gfarm:abc', 'dir1']
        files = ['gfarm:abc', 'dir1/readonly/file#2']

        # Gfarm -> Gfarm(tar)
        self.cmd_create(test1_dir_gfarm, srcdir_gfarm, members)
        # Gfarm(tar) -> Gfarm
        self.cmd_extract(test2_dir_gfarm, test1_dir_gfarm, files)

        # Local -> Local(tar)
        self.cmd_create(test3_dir_local, srcdir_local, files)
        # Local(tar) -> Local
        self.cmd_extract(test4_dir_local, test3_dir_local, files)
        self.cmd_extract(test5_dir_local, test3_dir_local, members)

        # error
        try:
            self.cmd_extract(test6_dir_local, test3_dir_local, ['abcde'])
            logger.error_exit(1, testname + '(not found in tar files) ' +
                              '... FAIL (unexpected success)')
        except GfptarError as e:
            logger.info(f"Expected error, ignored: {str(e)}")

        # check mode=0o700 for parents without tarinfo of file#2
        gd1 = GfURL.init(os.path.join(test2_dir_gfarm, 'dir1/readonly'))
        ld1 = GfURL.init(os.path.join(test4_dir_local, 'dir1/readonly'))
        gd1_mode = gd1.mode()
        if gd1_mode != 0o700:
            logger.error_exit(
                1, f"{testname} ... FAIL: gd1.mode={gd1_mode:#o}")
        ld1_mode = ld1.mode()
        if ld1_mode != 0o700:
            logger.error_exit(
                1, f"{testname} ... FAIL: ld1.mode={ld1_mode:#o}")

        for f in files:
            g_member = GfURL.init(os.path.join(test2_dir_gfarm, f))
            l_member = GfURL.init(os.path.join(test4_dir_local, f))
            l_member2 = GfURL.init(os.path.join(test5_dir_local, f))
            if not g_member.compare_data(l_member):
                logger.error_exit(1, testname
                                  + '... FAIL (data mismatch 1)')
            if not l_member.compare_data(l_member2):
                logger.error_exit(1, testname
                                  + '... FAIL (data mismatch 2)')
        print(testname + ' ... PASS')
        workdir_local_url.remove_tree(remove_readonly=True)
        workdir_gfarm_url.remove_tree(remove_readonly=True)

    def test_update_append_exclude(self):
        testname = 'gfptar-test-update-append-exclude'
        self.test_workdir_set(testname)
        # workdir_local = self.workdir_local
        # workdir_gfarm = self.workdir_gfarm
        workdir_local_url = self.workdir_local_url
        workdir_gfarm_url = self.workdir_gfarm_url

        testsrc_name1 = 'test-src1'
        testsrc_name2 = 'test-src2'
        testsrc_name3 = 'test-src3'
        srcdir_local1 = workdir_local_url.url_join(testsrc_name1)
        srcdir_local2 = workdir_local_url.url_join(testsrc_name2)
        srcdir_local3 = workdir_local_url.url_join(testsrc_name3)
        srcdir_gfarm1 = workdir_gfarm_url.url_join(testsrc_name1)
        srcdir_gfarm2 = workdir_gfarm_url.url_join(testsrc_name2)
        srcdir_gfarm3 = workdir_gfarm_url.url_join(testsrc_name3)
        expect_dir_name = 'test-expect'
        expect_dir_local = workdir_local_url.url_join(expect_dir_name)

        exclude_name = 'exclude.txt'
        exclude_file_l = workdir_local_url.url_join(exclude_name)
        exclude_file_l_url = GfURL.init(exclude_file_l)
        exclude_file_g = workdir_gfarm_url.url_join(exclude_name)
        exclude_file_g_url = GfURL.init(exclude_file_g)

        # for cmd_create
        tree = [
            (self.D, 'dir1', 0o715, 10, None, True),
            (self.F, 'dir1/file1', 0o644, 10, None, True),
            (self.F, 'dir1/file2', 0o664, 10, None, True),
            (self.S, 'dir1/symlink1', 0o777, 10, 'file1', True),
            (self.F, 'dir1/exclude1.ext1', 0o660, 10, None, True),
        ]
        self.test_create_tree(srcdir_local1, tree)
        self.test_create_tree(srcdir_gfarm1, tree)

        # for cmd_update
        tree = [
            (self.D, 'dir1', 0o750, 11, None, True),
            (self.F, 'dir1/file1', 0o444, 11, None, True),
            (self.F, 'dir1/file2', 0o664, 9, None, True),  # not updated
            (self.S, 'dir1/symlink1', 0o770, 11, 'file1', True),
            (self.F, 'dir1/exclude2.ext2', 0o660, 10, None, True),
            (self.F, 'dir1/exclude2.ext3', 0o660, 10, None, True),
        ]
        self.test_create_tree(srcdir_local2, tree)
        self.test_create_tree(srcdir_gfarm2, tree)

        # for cmd_append
        tree = [
            (self.D, 'dir2', 0o700, 10, None, True),
            (self.F, 'dir2/file3', 0o660, 100, None, True),
            (self.F, 'dir2/exclude4.ext4', 0o660, 10, None, True),
            (self.F, 'dir2/exclude5.ext5', 0o660, 10, None, True),
            (self.F, 'dir2/exclude6.ext6', 0o660, 10, None, True),
        ]
        self.test_create_tree(srcdir_local3, tree)
        self.test_create_tree(srcdir_gfarm3, tree)

        # expected tree
        tree = [
            (self.D, 'dir1', 0o750, 11, None, True),
            (self.F, 'dir1/file1', 0o444, 11, None, True),
            (self.F, 'dir1/file2', 0o664, 10, None, True),
            (self.S, 'dir1/symlink1', 0o770, 11, 'file1', True),
            (self.D, 'dir2', 0o700, 10, None, True),
            (self.F, 'dir2/file3', 0o660, 100, None, True),
        ]
        self.test_create_tree(expect_dir_local, tree)

        test1_name = 'test-1-create-l'
        test1_dir_local = workdir_local_url.url_join(test1_name)
        test2_name = 'test-2-extract-l'
        test2_dir_local = workdir_local_url.url_join(test2_name)

        test3_name = 'test-3-create-g'
        test3_dir_gfarm = workdir_gfarm_url.url_join(test3_name)
        test4_name = 'test-4-extract-g'
        test4_dir_local = workdir_local_url.url_join(test4_name)

        members1 = ['dir1']
        members2 = ['dir2']  # for cmd_append
        members_all = members1 + members2

        save_opt_exclude = self.opt['--exclude']
        save_opt_exclude_from = self.opt['--exclude-from']

        # for cmd_update
        with exclude_file_l_url.writeopen(textmode=True) as f:
            f.write('*.ext2\n')
            f.write('*.ext3\n')
        with exclude_file_g_url.writeopen(textmode=True) as f:
            f.write('*.ext2\n')
            f.write('*.ext3\n')

        # Local -> Local(tar)
        self.opt['--exclude'] = ['*.ext1']
        self.opt['--exclude-from'] = None
        self.cmd_create(test1_dir_local, srcdir_local1, members1)
        # Local -> Local(tar) (update)
        self.opt['--exclude'] = None
        self.opt['--exclude-from'] = exclude_file_l
        self.cmd_update(test1_dir_local, srcdir_local2, members1)
        # Local -> Local(tar) (append)
        self.opt['--exclude'] = ['*.ext4', '*.ext5']
        self.opt['--exclude-from'] = None
        self.cmd_append(test1_dir_local, srcdir_local3, members2)
        # Local(tar) -> Local
        self.opt['--exclude'] = ['*.ext6']
        self.opt['--exclude-from'] = None
        self.cmd_extract(test2_dir_local, test1_dir_local, members_all)

        # self.test_mode = False
        # self.cmd_list_verbose(test1_dir_local)
        # self.test_mode = True

        if not self.test_compare_local(test2_dir_local, expect_dir_local,
                                       same_owner=True):
            logger.error_exit(1, testname + ' ... FAIL (different data)(1)')

        # Gfarm -> Gfarm(tar)
        self.opt['--exclude'] = ['*.ext1']
        self.opt['--exclude-from'] = None
        self.cmd_create(test3_dir_gfarm, srcdir_gfarm1, members1)
        # Gfarm -> Gfarm(tar) (update)
        self.opt['--exclude'] = None
        self.opt['--exclude-from'] = exclude_file_g
        self.cmd_update(test3_dir_gfarm, srcdir_gfarm2, members1)
        # Gfarm -> Gfarm(tar) (append)
        self.opt['--exclude'] = ['*.ext4', '*.ext5']
        self.opt['--exclude-from'] = None
        self.cmd_append(test3_dir_gfarm, srcdir_gfarm3, members2)
        # Gfarm(tar) -> Local
        self.opt['--exclude'] = ['*.ext6']
        self.opt['--exclude-from'] = None
        # self.cmd_extract(test4_dir_local, test3_dir_gfarm, members_all)
        self.cmd_extract(test4_dir_local, test3_dir_gfarm, [])

        if not self.test_compare_local(test4_dir_local, expect_dir_local,
                                       same_owner=True):
            logger.error_exit(1, testname + ' ... FAIL (different data)(2)')

        print(testname + ' ... PASS')
        workdir_local_url.remove_tree(remove_readonly=True)
        workdir_gfarm_url.remove_tree(remove_readonly=True)
        self.opt['--exclude'] = save_opt_exclude
        self.opt['--exclude-from'] = save_opt_exclude_from

    # for test
    F = 'file'
    D = 'directory'
    S = 'symlink'
    L = 'hardlink'

    def test_prepare_srcdir(self, dir_url_str,
                            readonly=False, noread=False,
                            link=False, longname=False):
        logger.debug('readonly=%s, noread=%s, link=%s, longname=%s',
                     readonly, noread, link, longname)
        longname = ('0123456789' * 30)[:255]
        tree = [
            # (self.F, 'file1', 0o664, 1234567890, None, True),
            (self.F, 'gfarm:abc', 0o644, 1234567890,
             None, True),  # not Gfarm URL
            (self.D, 'dir1', 0o715, 2234567890, None, True),
        ]
        tree_readonly = [
            (self.D, 'dir1/readonly', 0o700,  # chmod 0o500 later
             3234567891, None, True),
            (self.F, 'dir1/readonly/file#2', 0o456, 4234567892, None, True),
        ]
        tree_noread = [
            (self.F, 'dir1/noread-file', 0o000, 1234567891, None, False),
            (self.D, 'dir1/noread-dir', 0o700,  # chmod 0o000 later
             1234567892, None, True),
            (self.F, 'dir1/noread-dir/file3', 0o400, 1234567893, None, False),
        ]
        tree_link = [
            (self.D, 'dir1/ディレクトリ 2', 0o755, 5234567890, None, True),
            (self.L, 'dir1/ディレクトリ 2/hardlink1',
             0o400, 6234567890, 'gfarm:abc', True),
            (self.S, 'dir1/ディレクトリ 2/symlink1',
             0o777, 7234567890, 'hardlink1', True),
        ]
        tree_longname = [
            (self.F, 'dir1/' + longname, 0o775, 9234567890, None, True),
        ]
        if readonly:
            tree += tree_readonly
        if noread:
            tree += tree_noread
        if link:
            tree += tree_link
        if longname:
            tree += tree_longname

        count = self.test_create_tree(dir_url_str, tree)
        if readonly:
            url_str = os.path.join(dir_url_str, 'dir1/readonly')
            readonly_url = GfURL.init(url_str)
            readonly_url.chmod(mode=0o500)
        if noread:
            url_str = os.path.join(dir_url_str, 'dir1/noread-dir')
            noread_url = GfURL.init(url_str)
            noread_url.chmod(mode=0o000)
        return count

    def test_create_tree(self, dir_url_str, tree):
        count = 0
        srcdir_url = GfURL.init(dir_url_str)
        srcdir_url.mkdir()
        count += 1
        for ent in tree:
            ftype = ent[0]
            path = ent[1]
            mode = ent[2]
            mtime = ent[3]
            linkname = ent[4]
            readable = ent[5]
            url = GfURL.init(srcdir_url.url_join(path))
            if ftype == self.F:
                with url.writeopen(textmode=True,
                                   mode=mode,
                                   mtime=mtime,
                                   use_fsync=False) as f:
                    f.write(path + str(mtime))
            elif ftype == self.D:
                url.mkdir()
            elif ftype == self.S:
                url.symlink(linkname)
                url.chmod(mode, mtime=mtime, follow_symlinks=False)
            elif ftype == self.L:
                url.hardlink(srcdir_url.url_join(linkname))
                url.chmod(mode, mtime=mtime)
            else:
                raise Exception(f"Unexpected ftype={ftype}")
            if readable:
                count += 1
        for ent in reversed(tree):
            ftype = ent[0]
            path = ent[1]
            mode = ent[2]
            mtime = ent[3]
            # linkname = ent[4]
            url = GfURL.init(srcdir_url.url_join(path))
            if ftype == self.D:
                url.chmod(mode, mtime=mtime)
        srcdir_url.chmod(mode=0o700, mtime=0)
        return count  # the number of readable files

    def test_specified_dir(self):
        basedir = self.opt['--basedir']
        if basedir is None:
            return
        infiles = self.opt['MEMBER']
        if len(infiles) == 0:
            return

        basedir_url = GfURL.init(basedir)

        testname = 'gfptar-test-specified-dir'
        self.test_workdir_set(testname)
        # workdir_local = self.workdir_local
        # workdir_gfarm = self.workdir_gfarm
        workdir_local_url = self.workdir_local_url
        workdir_gfarm_url = self.workdir_gfarm_url

        test9_name = 'test-9-create'
        test9_dir_gfarm = workdir_gfarm_url.url_join(test9_name)
        test10_name = 'test-10-extract'
        test10_dir_gfarm = workdir_gfarm_url.url_join(test10_name)
        test11_name = 'test-11-create'
        test11_dir_local = workdir_local_url.url_join(test11_name)
        test12_name = 'test-12-extract'
        test12_dir_local = workdir_local_url.url_join(test12_name)

        # basedir -> Gfarm(tar)
        self.cmd_create(test9_dir_gfarm, basedir, infiles)
        # Gfarm(tar) -> Gfarm
        self.cmd_extract(test10_dir_gfarm, test9_dir_gfarm, [])
        # Gfarm -> Local(tar)
        self.cmd_create(test11_dir_local, test10_dir_gfarm, infiles)
        # Local(tar) -> Local
        self.cmd_extract(test12_dir_local, test11_dir_local, [])

        result = True
        for infile in infiles:
            src = basedir_url.url_join(infile)
            result_dir = os.path.join(test12_dir_local, infile)
            # owner on local and Gfarm may be different
            same_owner = not basedir_url.is_gfarm()
            if not self.test_compare(result_dir, src, data=True,
                                     same_owner=same_owner):
                result = False
                break
        if result:
            print(testname + ' ... PASS')
            workdir_local_url.remove_tree(remove_readonly=True)
            workdir_gfarm_url.remove_tree(remove_readonly=True)
        else:
            logger.error_exit(1, testname + ' ... FAIL (different data)')

    def test_compare_local(self, dir1, dir2, data=False, same_owner=False):
        # SEE ALSO: test_prepare_srcdir
        exclude_list = [r'noread-*']
        diff_args = ['diff']
        for exclude in exclude_list:
            diff_args += ['--exclude', exclude]
        diff_args += ['-r', dir1, dir2]
        out, err, ret = execcmd_raw(diff_args)
        if ret != 0:
            logger.error(f"diff -r {dir1} {dir2}: {out}: {err}")
            return False
        logger.debug('diff -r (data check): PASS')
        return self.test_compare(dir1, dir2, data=data, same_owner=same_owner,
                                 exclude_list=exclude_list)

    def test_compare(self, dir1, dir2, data=True, same_owner=False,
                     exclude_list=None):
        # compare mtime and mode of dir1/files with dir2/files
        d1url = GfURL.init(dir1)
        d2url = GfURL.init(dir2)
        return d1url.compare_entries(d2url, data=True, same_owner=same_owner,
                                     exclude_list=exclude_list)

    def MT_enabled(self):
        # jobs == 0: not use multi-thread
        return self.jobs >= 1

    # lock required
    def info(self, fmt, *args):
        if self.verbose:
            print(fmt.format(*args))
        if self.debug:
            logger.debug(fmt.format(*args))

    def print_trace(self, exc):
        if self.verbose or self.debug:
            tb = traceback.TracebackException.from_exception(exc)
            logger.info(''.join(tb.format()))

    def list_dummy_files(self, base_dir, num, size_min, size_max):
        # defaults
        files_per_dir = 1000
        dir_min_depth = 5
        dir_max_depth = 5
        dir_min_length = 30
        dir_max_length = 30
        file_min_length = 30
        file_max_length = 30

        choices = string.ascii_letters + string.digits + '漢あア()[]-'
        other_symbols = ' !"#$%&=^~|`@{}+*;:,.<>?_' + r"\'"
        choices += other_symbols
        now = time.time()

        def generate_random_dirname():
            depth = random.randint(dir_min_depth, dir_max_depth)
            directories = [''.join(
                random.choices(choices,
                               k=random.randint(
                                   dir_min_length, dir_max_length)))
                           for _ in range(depth)]
            return directories

        def generate_random_filename():
            suffix = ''.join(random.choices(string.ascii_lowercase,
                                            k=random.randint(2, 4)))
            return ''.join(
                random.choices(choices, k=random.randint(
                    file_min_length, file_max_length))) + '.' + suffix

        def rand_dir(first_dir_index):
            first_dir = f"dir{first_dir_index:04d}"

            mode = 0o755
            file_type = GfURLEntry.TYPE_DIR
            uname = 'testuser1'
            gname = 'testgroup2'
            size = 0
            mtime = now
            linkname = ''
            # first_dir: to pick up members
            #
            # path = os.path.join(base_dir, first_dir)
            # yield GfURLEntry(path, mode, file_type, uname, gname,
            #                  size, mtime, linkname)
            dir_list = generate_random_dirname()
            dir_depth = len(dir_list)
            for i in range(dir_depth):
                parent = dir_list[:(i+1)]
                dir_path = os.path.join(*parent)
                path = os.path.join(base_dir, first_dir, dir_path)
                yield GfURLEntry(path, mode, file_type, uname, gname,
                                 size, mtime, linkname)

        def rand_file(dir_path, idx):
            # f = generate_random_filename()
            f = f"{idx}.txt"
            path = os.path.join(dir_path, f)
            mode = 0o600
            file_type = GfURLEntry.TYPE_FILE
            uname = 'testuser1'
            gname = 'testgroup1'
            if size_max > 0:
                size = random.randint(size_min, size_max)
            else:
                size = 0
            mtime = now
            linkname = ''
            return GfURLEntry(path, mode, file_type, uname, gname,
                              size, mtime, linkname)

        dir_num = int(num / files_per_dir)
        remainder = num % files_per_dir
        for dir_idx in range(dir_num):
            dir_count = 0
            for dir_ent in rand_dir(dir_idx):
                dir_count += 1
                yield dir_ent
            for j in range(files_per_dir - dir_count):
                yield rand_file(dir_ent.path, j)
        dir_idx = dir_num
        if remainder > 0:
            dir_count = 0
            for dir_ent in rand_dir(dir_idx):
                dir_count += 1
                yield dir_ent
            for j in range(remainder - dir_count):
                yield rand_file(dir_ent.path, j)

    TABLE_tar_entry = 'tar_entry'

    def cmd_create_init(self, cmd_name, outdir, basedir, infiles):
        self.cmd_name = cmd_name
        self.options_init()
        logger.debug('create start: outdir=%s, basedir=%s', outdir, basedir)
        self.outdir = outdir
        self.outdir_url = GfURL.init(outdir, dry_run=self.dry_run)
        self.basedir_url = GfURL.init(basedir)
        self.assumed_size = self.opt['--size']
        if self.assumed_size <= 0:
            self.assumed_size = 1
        self.max_entries_per_tar = self.opt['--max-entries-per-tar']
        if self.max_entries_per_tar <= 0:
            self.max_entries_per_tar = 1
        self.ratio = self.opt['--ratio']

        self.disable_gfarm_command = self.opt['--disable-gfarm-command']
        self.use_gfarm_command = not self.disable_gfarm_command
        self.gfsched_interval = self.opt['--gfsched-interval']

        self.dummy_num = self.opt['--dummy-num']
        if self.dummy_num is not None and self.dummy_num > 0:
            self.dummy_input = True
        else:
            self.dummy_input = False
        self.dummy_size_min = self.opt['--dummy-size-min']
        self.dummy_size_max = self.opt['--dummy-size-max']

        self.archived_size = 0
        self.stored_size = 0
        self.stored_num = 0
        self.selected_size = 0
        self.selected_num = 0
        self.total_size = 0
        self.total_num = 0
        self.start_time = time.time()
        self.next_time = self.start_time + self.progress_interval
        self.listing = True

        self.gfsched_init()

        self.cannot_be_archived = 0
        self.create_job_init()  # before starting threads
        self.fattr_dict_list = []  # to compare mtime for --update

        self.infiles_checked = set()
        for infile in infiles:
            # infile_url = GfURL.init(infile)
            # if not infile_url.is_local():
            #     raise GfException('specifying a relative path is required '
            #                       'instead of a URL: ' + infile)
            # NOTE: allow members named gfarm:* (ex. gfarm:/tmp/gfarm:abc)
            infile_url = GfURL.init(infile, local=True)
            infile = infile_url.path  # normalize and ignore scheme
            # normalized: ex. .///abc -> ./abc
            infile = infile.lstrip('/')  # relative path only
            if infile.startswith('./'):
                infile = infile[2:]
            if infile == '' or infile == '.':
                raise GfException('specifying current directory (.) '
                                  + 'as members is not allowed: ' + infile)
            if infile == '..' or infile.startswith('../'):
                raise GfException('specifying parent directory (..) '
                                  + 'as members is not allowed: ' + infile)

            url_str = os.path.join(self.basedir_url.url_str, infile)
            url = GfURL.init(url_str)
            if not url.exists():
                raise FileNotFoundError(url_str)
            self.infiles_checked.add(infile)

    def cmd_create(self, outdir, basedir, infiles):
        self.cmd_create_init('create', outdir, basedir, infiles)
        self.outdir_url.create_new_dir()
        serial = 1
        gen = 1
        self.cmd_create_common(serial, gen)

    def error_type_mismatch(self, suffix, tar_path):
        return GfptarError(f"Compression type mismatch ({suffix}): {tar_path}")

    def suffix_check(self, suffix, tar_path):
        if suffix is None:
            suffix = os.path.splitext(tar_path)[1]
            dottype = suffix.split('.')
            if len(dottype) == 2:
                self.compress_type = dottype[1]
                if self.compress_type == 'tar':
                    self.compress_type = Compress.TYPE_NO
            if self.verbose:
                self.progress_newline()
            logger.info(f"recognized type is {self.compress_type}")
            Compress.init(self.compress_type, self.compress_prog,
                          self.gzip_prog, self.bzip2_prog, self.xz_prog)
        elif not tar_path.endswith(suffix):
            raise self.error_type_mismatch(suffix, tar_path)
        return suffix

    def cmd_append(self, outdir, basedir, infiles):
        self.cmd_create_init('append', outdir, basedir, infiles)
        if not self.outdir_url.exists():
            raise FileNotFoundError(outdir)
        max_serial = 0
        max_gen = 0
        suffix = self.compress_type_specified  # may be None
        tar_list, dbgz_list = self.list_tar_list_dbgz_from_dir(self.outdir_url)
        for serial, gen, tar_path in tar_list:
            if serial > max_serial:
                max_serial = serial
            if gen > max_gen:
                max_gen = gen
            suffix = self.suffix_check(suffix, tar_path)
        self.cmd_create_common(max_serial + 1, max_gen + 1)

    def cmd_update(self, outdir, basedir, infiles):
        self.cmd_create_init('update', outdir, basedir, infiles)
        if not self.outdir_url.exists():
            raise FileNotFoundError(outdir)

        # To open all DB files at the same time
        update_max_openfiles()

        broken_count = 0
        max_serial = 0
        max_gen = 0
        suffix = self.compress_type_specified  # may be None
        fattr_dict = None

        mtime_db_idx = 0
        num_entries = 0
        if self.enable_tempdb_aggregation:
            MAX_ENTRIES = self.max_tempdb_aggregation
        else:
            MAX_ENTRIES = -1

        for serial, gen, tar_path, dbgz_path, db_path in \
                self.list_tar_dbgz_db(self.outdir_url, sort=True,
                                      progress=self.progress_enabled):
            if self.is_canceled():
                raise self.error_canceled()
            if serial > max_serial:
                max_serial = serial
            if gen > max_gen:
                max_gen = gen
            if dbgz_path is None or tar_path is None:
                broken_count += 1
                continue
            suffix = self.suffix_check(suffix, tar_path)
            if MAX_ENTRIES >= 0:
                if fattr_dict is None or \
                   (MAX_ENTRIES != 0 and num_entries >= MAX_ENTRIES):
                    tmpdb_path = os.path.join(self.tmpdir.name,
                                              f"mtime-{mtime_db_idx}.db")
                    db = DB(tmpdb_path)
                    fattr_dict = DBDict(db, Mtime, 'mtime')
                    self.fattr_dict_list.append(fattr_dict)
                    mtime_db_idx += 1
                    num_entries = 0
                for path, fattr in InfoDB.list_entries_from_db(
                        db_path, resolve_ugmap=False, sort=None):
                    if self.is_canceled():
                        raise self.error_canceled()
                    fattr_dict[path] = Mtime(fattr.mtime)
                    num_entries += 1
                logger.debug('os.remove: %s', db_path)
                os.remove(db_path)
            else:
                db = DB(db_path)
                fattr_dict = DBDict(db, FileAttr2, InfoDB.TABLE_ENTRY)
                self.fattr_dict_list.append(fattr_dict)

        self.start_time = time.time()
        self.next_time = self.start_time + self.progress_interval
        self.cmd_create_common(max_serial + 1, max_gen + 1)
        if broken_count > 0:
            raise GfptarError("Some .tar.* or .db.gz are broken:"
                              f" count={broken_count}")

    def cmd_create_common(self, serial, gen):
        if self.compress_type == Compress.TYPE_NO:
            self.split_size = self.assumed_size
            self.suffix = '.tar'
        else:
            self.split_size = self.assumed_size * 100 / self.ratio
            self.suffix = '.tar.' + self.compress_type

        def listdir_switch(gfurl):
            if self.dummy_num is not None:
                return self.list_dummy_files(gfurl.url_str,
                                             self.dummy_num,
                                             self.dummy_size_min,
                                             self.dummy_size_max)
            else:
                return gfurl.listdir(recursive=True, first=True,
                                     hardlink_warn=self.hardlink_warn)

        def is_update_target_true(entry):
            return True

        def is_update_target0(entry):
            # entry is an input file.
            path = entry.subpath(self.basedir_url)
            # compare mtime from newer .db
            for fattr_dict in reversed(self.fattr_dict_list):
                # fattr_dict has existing files.
                fattr = fattr_dict.get(path)
                if fattr is None:
                    continue
                elif int(entry.mtime) > int(fattr.mtime):  # sec.
                    logger.debug(f"is_update_target0: path={path}:"
                                 f" entry.mtime({entry.mtime}) >"
                                 f" fattr.mtime({fattr.mtime})")
                    return True
                # NOTE: compare only mtime
                # elif fattr.size != entry.size:
                #     logger.debug(f"is_update_target0: path={path}:"
                #                  f" fattr.size({fattr.size})"
                #                  f" != entry.size({entry.size})")
                #     return True
                else:
                    return False
            # not found
            logger.debug(f"is_update_target0: path={path}: not found (True)")
            return True

        if len(self.fattr_dict_list) > 0:
            is_update_target = is_update_target0
        else:
            is_update_target = is_update_target_true

        tmpdir = self.tmpdir
        tardb_prefix = os.path.join(tmpdir.name, self.cmd_name)
        tardb_fmt = f"_{self.SERIAL_FORMAT}.db"
        # to reduce memory usage
        filelist_db = DB(tardb_prefix + tardb_fmt % serial)
        filelist = DBList(filelist_db, GfURLEntry, self.TABLE_tar_entry)
        filelist_num = 0
        filelist_size = 0

        cannot_read_dir = 0
        has_error = None

        if self.progress_enabled:
            self.progress_for_create(self.start_time)

        for infile in self.infiles_checked:
            if self.is_canceled():
                logger.debug('Canceled (listdir 1): serial=%d', serial)
                break

            url_str = os.path.join(self.basedir_url.url_str, infile)
            gfurl = GfURL.init(url_str)
            logger.debug('listdir: %s', gfurl.url_str)
            try:
                for entry in listdir_switch(gfurl):
                    if self.is_canceled():
                        logger.debug('Canceled (listdir 2): serial=%d', serial)
                        break

                    if self.progress_enabled:
                        now = time.time()
                        with self.lock():
                            if now >= self.next_time:
                                self.next_time = now + self.progress_interval
                                self.progress_for_create(now)

                    logger.debug('listdir: entry.path=%s', entry.path)
                    # include length of path
                    this_size = entry.size_all()
                    with self.lock():  # for progress
                        self.total_size += this_size
                        self.total_num += 1

                    if self.exclude_match(entry.path):
                        logger.info(f"Not added (excluded): {entry.path}")
                        continue  # skip
                    if not is_update_target(entry):
                        logger.info(f"Not added (not new): {entry.path}")
                        continue  # skip

                    with self.lock():  # for progress
                        self.selected_size += this_size
                        self.selected_num += 1

                    if filelist_num > 0 \
                       and (filelist_size + this_size > self.split_size
                            or filelist_num + 1 > self.max_entries_per_tar):
                        try:
                            filelist.commit()
                            filelist.close()
                            # DO NOT share a DBList with children,
                            # share the filename instead.
                            self.create_job_execute(gen, serial,
                                                    filelist.filename())
                        except Exception as e1:
                            self.cancel()
                            if has_error is None:
                                has_error = e1
                        serial += 1
                        filelist_db = DB(tardb_prefix + tardb_fmt % serial)
                        filelist = DBList(filelist_db, GfURLEntry,
                                          self.TABLE_tar_entry)
                        filelist_num = 0
                        filelist_size = 0
                    filelist.append(entry)
                    filelist_num += 1
                    filelist_size += this_size

                    if has_error is not None:
                        break  # from listdir_switch()
            except MemoryError as e2:
                self.cancel()
                has_error = e2
                filelist = []
                filelist_num = 0
                filelist_size = 0
                break  # from loop of infiles_checked
            except Exception as e2:
                cannot_read_dir += 1
                self.progress_newline()
                logger.warning('%s: error while reading directory (%s)',
                               gfurl.url_str, convert_message(e2))
                self.print_trace(e2)
                # continue (next infile)

        if has_error is not None:
            self.cancel()

        if filelist_num > 0 and not self.is_canceled():
            try:
                filelist.commit()
                filelist.close()
                self.create_job_execute(gen, serial, filelist.filename())
            except Exception as e:
                self.cancel()
                if has_error is None:
                    has_error = e

        if not self.is_canceled():
            with self.lock():  # for progress
                self.listing = False

        self.create_job_final()

        for fattr_dict in self.fattr_dict_list:
            fattr_dict.close()
        tmpdir.cleanup()

        if self.progress_enabled:
            self.progress_for_create(time.time())
            self.progress_newline()
        if self.cannot_be_archived > 0 or cannot_read_dir > 0:
            e = CannotBeArchivedError(self.cannot_be_archived, cannot_read_dir)
            logger.warning(str(e))
            if has_error is None:
                has_error = e
        if not self.quiet and self.stored_size > 0:
            print('compression ratio: %.2f%% (%d/%d)' %
                  (100 * self.archived_size / self.stored_size,
                   self.archived_size, self.stored_size))
        if self.selected_num == 0 and self.cmd_name == 'update':
            print('No files were updated.')
        if has_error is not None:
            raise has_error
        if self.is_canceled():
            raise self.error_canceled()

    def create_job_init(self):
        if self.MT_enabled():
            self.lock_init(True)
            self.create_job_execute = self._create_job_execute_MT
            self.thread_pool = concurrent.futures.ThreadPoolExecutor(
                max_workers=self.max_workers)
            self.futures = {}  # key: serial number

        else:
            self.lock_init(False)
            self.create_job_execute = self._create_job_execute

        # kill old processes for regress
        self.worker_terminate()

        class Started():
            pass

        # create Process()es before starting threads
        self.worker_ident_dict = {}
        for i in range(self.max_workers):
            inq = multiprocessing.Queue()
            outq = multiprocessing.Queue()
            # NOTE: use multiprocessing.Process instead of ProcessPoolExecutor.
            # ProcessPoolExecutor cannot be utilized here,
            # because, for example, multiprocessing.Queue() cannot be
            # specified as arguments for ProcessPoolExecutor.
            # multiprocessing.Manager().Queue() can be used
            # for ProcessPoolExecutor, but it is very slow.
            started = Started()
            process = multiprocessing.Process(target=self.create_a_tar_process,
                                              args=(inq, outq))
            started.status = False  # lock required
            self.worker_list.append((started, process, inq, outq))

    def _create_job_execute(self, gen, serial, arg):
        self.create_a_tar_thread(gen, serial, arg)

    def _create_job_execute_MT(self, gen, serial, arg):
        t = self.thread_pool.submit(self.create_a_tar_thread, gen, serial, arg)
        self.futures[t] = serial
        has_error = None
        try:
            has_error = self._create_job_check_MT()
        except MemoryError:
            raise
        except Exception as e:
            # NOTE: cannot catch TimeoutError
            # ignore timeout
            logger.debug('_create_job_execute_MT()', exc_info=e)
        if has_error is not None:
            raise has_error

    def process_close(self, process, index=None):
        if not process.is_alive():
            return
        logger.debug('process_close() start (index=%s): pid=%s',
                     index, process.pid)
        # process.terminate()
        os.kill(process.pid, signal.SIGKILL)
        process.join()
        logger.debug('process_close() finished (index=%d)', index)

    def create_job_final(self, timeout=None):
        if self.MT_enabled():
            has_error = self._create_job_check_MT(timeout=timeout)
            self.thread_pool.shutdown(wait=False)
            logger.debug('self.thread_pool.shutdown')
            if has_error is not None:
                raise has_error
        self.worker_terminate()

    def worker_terminate(self):
        for index, worker in enumerate(self.worker_list):
            started, process, inq, outq = worker
            # NOTE: processes will be killed.
            # if started.status:
            #     with ignore_exception(True):
            #         inq.put(self.MSG_PARENT_EXIT)
            with ignore_exception(True):
                inq.close()
            with ignore_exception(True):
                outq.close()
            try:
                if started.status:
                    self.process_close(process, index=index)
                started.status = False
            except Exception as e:
                logger.debug("self.process_close", exc_info=e)
                # ignore
        self.worker_list = []

    def _create_job_check_MT(self, timeout=0.1):
        has_error = None
        for t in concurrent.futures.as_completed(self.futures,
                                                 timeout=timeout):
            exc = t.exception()
            if exc:
                logger.debug('serial=%d: %s', self.futures[t],
                             convert_message(exc))
                self.print_trace(exc)
                if not self.is_canceled():
                    self.cancel()
                    # DO NOT cancel threads because it will freeze.
                    # for t2 in self.futures:
                    #     if t == t2:
                    #         continue
                    #     logger.error('Canceled: %s', self.futures[t2])
                    #     t2.cancel()
                if has_error is None:  # save first error
                    has_error = exc
        return has_error

    def lock_init(self, enable, timeout=None):
        if enable:
            self._lock = threading.Lock()
            self.lock = self.lock_timeout
            if timeout is not None:
                self._lock_timeout = timeout
            else:
                self._lock_timeout = -1
        else:
            self._lock = None
            self.lock = self.lock_noop

    @contextmanager
    def lock_noop(self):
        yield

    @contextmanager
    def lock_timeout(self):
        locked = True
        if not self._lock.acquire(timeout=self._lock_timeout):
            logger.warning('lock: timeout(unexpected)')
            locked = False
        try:
            yield
        finally:
            if locked:
                self._lock.release()

    def clear_canceled(self):
        self.canceled = False

    def cancel(self):
        self.canceled = True

    def is_canceled(self):
        return self.canceled

    def gfsched_init(self):
        self.gfsched_lock = threading.Lock()
        self.gfsched_next = 0
        self.gfsched_list = []

    def _gfsched_sometimes(self, gfurl):
        # if gfurl.is_gfarm() is not True:
        #     raise AssertionError
        now = time.time()
        with self.gfsched_lock:
            if now < self.gfsched_next:
                return  # not updated
            # access to Gfarm
            old_schedlist = self.gfsched_list
            try:
                schedlist = gfurl.gfsched(write_mode=True,
                                          number=self.jobs)
            except Exception:
                logger.warning('Error occurred in gfsched')
                schedlist = old_schedlist
            self.gfsched_list = schedlist
            logger.debug("update gfsched_list")
            self.gfsched_next = now + self.gfsched_interval

    def select_a_target_host(self, outurl, index):
        if self.dry_run:
            return
        if not outurl.is_gfarm():
            return None
        if self.jobs <= 1:  # not use gfsched
            return None
        self._gfsched_sometimes(outurl)
        with self.gfsched_lock:
            schedlist_len = len(self.gfsched_list)
            if schedlist_len == 0:
                logger.warning(
                    'The destination host cannot be selected by gfptar')
                return None  # no candidate
            if schedlist_len > self.jobs:
                # use only top hosts
                self.gfsched_list = self.gfsched_list[:self.jobs]
                schedlist_len = len(self.gfsched_list)
            target_host = self.gfsched_list[index % schedlist_len]
        logger.debug("selected target_host: %s", target_host)
        return target_host

    def create_a_tar_thread(self, gen, serial, dbfile):
        try:
            self.create_a_tar_thread0(gen, serial, dbfile)
        except Exception as e:
            if self.is_canceled():
                raise self.error_canceled() from e
            else:
                raise

    MSG_CHILD_READY = 'READY'
    MSG_CHILD_PROGRESS = 'PROGRESS'
    MSG_CHILD_DONE = 'DONE'
    MSG_CHILD_ERROR = 'ERROR'
    MSG_PARENT_START = 'START'
    MSG_PARENT_CANCEL = 'CANCEL'
    MSG_PARENT_ERROR_COMPLETE = 'ERROR_COMPLETE'
    MSG_PARENT_EXIT = 'EXIT'

    def create_a_tar_thread0(self, gen, serial, arg):
        logger.debug('create_a_tar: start (gen=%d, serial=%d)', gen, serial)
        if self.is_canceled():
            logger.debug('Canceled (create 1): serial=%d', serial)
            return

        with self.lock():
            # Thread ID -> index of worker_list
            ident = threading.get_ident()
            idx = self.worker_ident_dict.get(ident)
            if idx is None:
                idx = len(self.worker_ident_dict)
                self.worker_ident_dict[ident] = idx
                logger.debug('worker_ident_dict: '
                             + str(self.worker_ident_dict))

        started, process, inq, outq = self.worker_list[idx]
        if not started.status:
            process.start()  # high cost to start
            started.status = True

        inq.put((self.MSG_PARENT_START, gen, serial, arg))
        response = outq.get()
        if response != self.MSG_CHILD_READY:
            logger.error(f"Unexpected response from child process: {response}")
            return

        try:
            cancel = False
            while True:
                if self.is_canceled() and not cancel:
                    logger.debug('Canceled (create 2): serial=%d', serial)
                    inq.put(self.MSG_PARENT_CANCEL)
                    cancel = True  # prevent input_queue from filling up
                try:
                    result = outq.get(timeout=1)
                except queue.Empty:
                    logger.debug('MSG_CHILD_ timeout, retry...')
                    result = None
                    if not process.is_alive():
                        logger.warning('Unexpected child process termination'
                                       ' (serial=%d)', serial)
                        break
                if result is None:
                    logger.debug('waiting for message from child process:'
                                 f" serial={serial}")
                    continue
                if len(result) == 0:
                    logger.warning('unknown result (None) from child process')
                    break
                msg = result[0]
                if msg == self.MSG_CHILD_PROGRESS:
                    logger.debug('MSG_CHILD_PROGRESS')
                    _msg, num, size, path_list = result
                    with self.lock():
                        for entry_path in path_list:
                            self.info('archived: {}', entry_path)
                        self.stored_size += size
                        self.stored_num += num
                        if self.progress_enabled:
                            now = time.time()
                            if now >= self.next_time:
                                self.next_time = now \
                                    + self.progress_interval
                                self.progress_for_create(now)
                elif msg == self.MSG_CHILD_DONE:
                    (_msg, tar_size, cannot_be_archived,
                     out_tar_path, out_dbgz_path) = result
                    with self.lock():
                        self.archived_size += tar_size
                        self.cannot_be_archived += cannot_be_archived
                        self.info('created({}): {}', self.suffix, out_tar_path)
                        self.info('created(.db.gz): {}', out_dbgz_path)
                    break
                elif msg == self.MSG_CHILD_ERROR:
                    (_msg, exc_type_name, exc_value_str,
                     exc_traceback_str) = result
                    inq.put(self.MSG_PARENT_ERROR_COMPLETE)
                    if self.verbose or self.debug:
                        raise GfException(
                            f"{exc_type_name}: {exc_value_str}\n"
                            f"{''.join(exc_traceback_str)}")
                    else:
                        raise GfException(f"{exc_type_name}: {exc_value_str}")
                else:
                    logger.error('Unexpected message from child process:'
                                 f" {result}")
                    break
        finally:
            logger.debug('(parent) subprocess finished: serial=%d', serial)

    def create_a_tar_process(self, input_queue, output_queue):
        global logger
        logger = logger_init_without_lock(
            self.name, loglevel=self.loglevel,
            debug=self.debug, verbose=self.verbose)

        pid = os.getpid()
        logger.debug('create_a_tar_process: start, pid=%d', pid)

        self.lock_init(False)  # child process
        self.sig_init(message=False)
        while self.create_a_tar_process0(input_queue, output_queue):
            pass
        logger.debug('create_a_tar_process: exit, pid=%d', pid)

    def create_a_tar_process0(self, input_queue, output_queue):
        while True:
            try:
                request = input_queue.get(timeout=1)
                break
            except queue.Empty:
                if self.is_canceled():  # SIGTERM from parent
                    return False  # exit
        logger.debug('create_a_tar_process0: request=%s', request)
        if request == self.MSG_PARENT_EXIT:  # not used
            return False  # exit
        elif request[0] == self.MSG_PARENT_START:
            output_queue.put(self.MSG_CHILD_READY)
            # OK
        else:
            raise GfException('Unexpected message from parent process:'
                              f" {request}")
        op, gen, serial, dbfile = request
        logger.debug('create_a_tar_process0: start (serial=%d)', serial)
        try:
            result = self.create_a_tar_process1(input_queue, output_queue,
                                                gen, serial, dbfile)
            tar_size, cannot_be_archived, outurl, infodb_url = result
            output_queue.put((self.MSG_CHILD_DONE, tar_size,
                              cannot_be_archived, outurl, infodb_url))
            logger.debug('subprocess exits: serial=%d', serial)
            return True  # next job
        except Exception as e:
            logger.debug('create_a_tar_process: error (serial=%d)', serial)
            self.print_trace(e)
            exc_type, exc_value, exc_traceback = sys.exc_info()
            exc_type_name = exc_type.__name__
            exc_value_str = str(exc_value)
            exc_traceback_str = traceback.format_exception(
                exc_type, exc_value, exc_traceback)
            output_queue.put((self.MSG_CHILD_ERROR,
                              exc_type_name, exc_value_str, exc_traceback_str))
            try:
                input_queue.get(timeout=10)
                # expect: self.MSG_PARENT_ERROR_COMPLETE
            except queue.Empty:
                pass
            return False  # exit

    def create_a_tar_process1(self, input_queue, output_queue,
                              gen, serial, dbfile):
        logger.debug('create_a_tar_process1: start (serial=%d)', serial)
        tardb = DB(dbfile)
        filelist = DBList(tardb, GfURLEntry, self.TABLE_tar_entry)

        if len(filelist) == 0:
            tardb.close()
            tardb.unlink()
            logger.warning(f"empty filelist: {dbfile}")
            return 0, 0, '', ''
        first = None
        last = None
        for entry in filelist:
            if entry.is_file():
                first = entry
                break
        if first is None:
            first = filelist[0]
        for entry in reversed(filelist):
            if entry.is_file():
                last = entry
                break
        if last is None or first.path == last.path:
            firstpath = first.subpath(self.basedir_url)
            outname = '%s%s' % (firstpath, self.suffix)
        else:
            firstpath = first.subpath(self.basedir_url)
            lastpath = last.subpath(self.basedir_url)
            outname = '%s..%s%s' % (firstpath, lastpath, self.suffix)

        serial_str = f"{self.SERIAL_FORMAT}_" % serial
        if gen >= 2:
            # SEE ALSO: InfoDB.dbgz_filename()
            prefix_str = f"g{gen}_{serial_str}"
        else:
            prefix_str = serial_str
        outname_max = self.outdir_url.MAXNAMLEN \
            - len(prefix_str) - len(self.TMP_SUFFIX)
        outname_len = len(outname.encode())
        offset = 0
        while outname_len > outname_max:
            # use last half of name
            outname = outname[-(outname_max-offset):]
            outname_len = len(outname.encode())
            logger.debug('modified outname_len=%d', outname_len)
            # loop for multibyte characters
            offset += 1
        # ex.: home/user1/dir -> home_user1_dir
        outname = prefix_str + outname.replace('/', '_')
        out_urlstr = self.outdir_url.url_join(outname)
        tmp_urlstr = out_urlstr + self.TMP_SUFFIX
        tmp_url = GfURL.init(tmp_urlstr,
                             use_gfarm_command=self.use_gfarm_command,
                             dry_run=self.dry_run)
        if tmp_url.is_gfarm():
            # direct access when outdir is gfarm2fs
            tmp_url_direct = GfURL.init(tmp_urlstr,
                                        use_gfarm_command=False,
                                        dry_run=self.dry_run)
        else:
            tmp_url_direct = tmp_url

        def remove(gfurl):
            if gfurl.exists():
                gfurl.remove()

        remove(tmp_url_direct)
        try:
            return self.create_a_tar_process2(gen, serial, tardb,
                                              input_queue, output_queue,
                                              tmp_url, tmp_url_direct,
                                              filelist, out_urlstr)
        except Exception:
            remove(tmp_url_direct)
            raise

    def create_a_tar_process2(self, gen, serial, tardb,
                              input_queue, output_queue,
                              tmp_url, tmp_url_direct,
                              filelist, out_urlstr):
        target_host = self.select_a_target_host(tmp_url, serial)
        tar_tmp = GfTarFile.create_open(tmp_url, self.bufsize,
                                        use_fsync=self.use_fsync,
                                        target_host=target_host,
                                        dummy_input=self.dummy_input)
        # to reduce memory usage
        # SEE ALSO: InfoDB.generate_db_and_dbgz()
        db_name = InfoDB.dbgz_filename(gen, serial)
        tmpdb_path = os.path.join(self.tmpdir.name, db_name + self.TMP_SUFFIX)
        infodb = InfoDB(tmpdb_path)

        cannot_be_archived = 0
        db_close = False
        cancel = False
        has_error = False
        next_time = time.time() + self.progress_interval
        progress_path_list = []
        progress_path_list_max = 100000  # limiter to limit memory usage
        progress_num = 0
        progress_size = 0
        report_path = (self.debug or self.verbose)
        try:
            for entry in filelist:
                # logger.debug('subprocess(serial=%d): %s', serial, entry.path)
                while not input_queue.empty():
                    try:
                        qdata = input_queue.get(timeout=1)
                    except queue.Empty:
                        qdata = None
                    if qdata == self.MSG_PARENT_CANCEL:
                        logger.debug('receive CANCEL from parent process')
                        cancel = True
                        break
                    else:
                        logger.error('unexpected message from parent process')
                        cancel = True
                        break
                if cancel or self.is_canceled():
                    break
                subpath = entry.subpath(self.basedir_url)
                try:
                    logger.debug('tar_tmp.add_entry: %s', subpath)
                    # GfTarFile.METHOD_add_entry
                    tarinfo = tar_tmp.add_entry(subpath, entry)
                    infodb.add(tarinfo)
                    progress_num += 1
                    progress_size += entry.size_all()
                    if report_path:
                        progress_path_list.append(subpath)
                    now = time.time()
                    if now >= next_time \
                       or len(progress_path_list) > progress_path_list_max:
                        next_time = now + self.progress_interval
                        output_queue.put((self.MSG_CHILD_PROGRESS,
                                          progress_num, progress_size,
                                          progress_path_list))
                        # reset
                        progress_path_list = []
                        progress_num = 0
                        progress_size = 0
                except MemoryError:
                    raise
                except BrokenPipeError:
                    raise
                except Exception as e:
                    cannot_be_archived += 1
                    logger.warning(convert_message(e))
                    continue
            # success
            if progress_num > 0:
                output_queue.put((self.MSG_CHILD_PROGRESS,
                                  progress_num, progress_size,
                                  progress_path_list))
            infodb.commit_close()
            db_close = True
            if cancel:
                raise self.error_canceled()
        except Exception:
            has_error = True
            raise
        finally:
            with ignore_exception(has_error):
                tar_tmp.close()
            with ignore_exception(has_error):
                tardb.close()
            with ignore_exception(has_error):
                tardb.unlink()
            with ignore_exception(has_error):
                if not db_close:
                    infodb.db.close()

        # success
        if self.dry_run:
            tar_size = 1024 ** 2  # no particular reason
        else:
            tar_size = tmp_url.size()
        tmp_url_direct.rename(out_urlstr)

        dbgz_urlstr = self.outdir_url.url_join(db_name)
        InfoDB.compress_copy(infodb.db.filename, dbgz_urlstr,
                             self.bufsize, self.use_fsync, self.dry_run)

        # for DEBUG
        # raise Exception('unexpected raise')

        return tar_size, cannot_be_archived, out_urlstr, dbgz_urlstr

    def error_canceled(self):
        return GfptarError('Canceled')

    def error_not_a_gfptar_directory(self, url_str):
        return GfptarError('Not a gfptar-archived directory: ' + url_str)

    def cmd_extract(self, outdir, indir, specified_members):
        self.options_init()
        logger.debug('extract start: outdir=%s, indir=%s', outdir, indir)
        self.outdir = outdir
        self.outdir_url = GfURL.init(outdir, dry_run=self.dry_run)
        self.outdir_url.create_new_dir()
        self.indir = indir
        self.specified_members = specified_members
        self.same_owner = self.opt['--same-owner']

        self.disable_gfarm_command = self.opt['--disable-gfarm-command']
        self.use_gfarm_command = not self.disable_gfarm_command
        self.gfsched_interval = self.opt['--gfsched-interval']

        self.indir_url = indir_url = GfURL.init(self.indir)
        if not indir_url.exists():
            raise FileNotFoundError(indir_url.url_str)
        if not indir_url.is_directory():
            raise self.error_not_a_gfptar_directory(indir_url.url_str)

        self.search_target = len(specified_members) > 0

        tmpdir = self.tmpdir
        db_file = os.path.join(tmpdir.name, 'extract.db')
        logger.debug('db_file=%s', db_file)

        # If outdir is gfarm2fs, the path will be converted into Gfarm URL.
        outdir_url_gf = GfURL.init(outdir, use_gfarm_command=True)
        if outdir_url_gf.is_local():
            self.use_fsync_for_x = False  # to improve performance
            self.outdir_is_local = True
        else:
            self.use_fsync_for_x = self.use_fsync
            self.outdir_is_local = False

        # to reduce memory usage
        self.db = DB(db_file, check_same_thread=False)

        self.extract_main()
        tmpdir.cleanup()

    TABLE_directory_set = 'directory_set'
    TABLE_member_set = 'member_set'
    TABLE_archive_dict = 'archive_dict'
    TABLE_member_check_dict = 'member_check_dict'
    TABLE_created_directory_set = 'created_directory_set'
    TABLE_dirstat_dict = 'dirstat_dict'

    def extract_schedule(self):
        directory_set = DBSet(self.db, StrObj, self.TABLE_directory_set)
        member_set = DBSet(self.db, StrObj, self.TABLE_member_set)
        archive_dict = DBDict(self.db, JsonObj, self.TABLE_archive_dict)
        member_check_dict = DBDict(self.db, JsonObj,
                                   self.TABLE_member_check_dict)

        self.created_directory_set = DBSet(self.db, StrObj,
                                           self.TABLE_created_directory_set)
        self.dirstat_dict = DBDict(self.db, FileAttr1, self.TABLE_dirstat_dict)

        if self.search_target:
            for member in self.specified_members:
                m = GfURL.init(member, local=True)  # relative path
                path = m.path  # normalized path
                if path.startswith('/'):
                    path = path.lstrip('/')
                member_check_dict[path] = False  # initialize

        gen_tarlist_dict = defaultdict(list)  # gen -> tarlist

        self.total_num = 0
        self.total_size = 0
        self.selected_num = 0
        self.selected_size = 0
        self.start_time = time.time()
        self.next_time = self.start_time + self.progress_interval
        if self.progress_enabled:
            self.progress_for_schedule(self.start_time)

        error_num = 0
        for serial, gen, tar_path, dbgz_path, db_path in \
                self.list_tar_dbgz_db(self.indir_url, sort=True,
                                      progress=False):
            if self.is_canceled():
                raise self.error_canceled()
            if dbgz_path is None or tar_path is None:
                error_num += 1
                continue
            for path, fattr in InfoDB.list_entries_from_db(
                    db_path, resolve_ugmap=False, sort=None):
                if self.is_canceled():
                    raise self.error_canceled()
                size = fattr.size
                file_type = fattr.ftype
                self.total_num += 1
                self.total_size += size
                if self.search_target:
                    logger.debug('archive_dict[%s]: %s', path, file_type)
                    archive_dict[path] = (file_type, size)
                else:
                    if file_type == InfoDB.TYPE_DIR:  # all directories
                        logger.debug('directory_set.add: %s', path)
                        directory_set.add(path.strip('/'))
            logger.debug('os.remove: %s', db_path)
            os.remove(db_path)

            if self.search_target:
                is_target = False
                for member in member_check_dict.keys():
                    ftype_size = archive_dict.get(member, None)
                    logger.debug('from member_check_dict.keys: %s [%s]',
                                 member, str(ftype_size))
                    found = False
                    if ftype_size is not None:
                        file_type, size = ftype_size
                        logger.debug('member_set.add: %s', member)
                        member_set.add(member)
                        self.selected_num += 1
                        self.selected_size += size
                        found = True
                        if file_type == InfoDB.TYPE_DIR:
                            logger.debug('directory_set.add: %s', member)
                            directory_set.add(member)
                            is_dir = True
                        else:
                            is_dir = False
                    else:
                        # specified member may be a parent directory
                        # ex. dir -> dir/* -> dir/path/to/file
                        is_dir = True
                    if is_dir:
                        # find {member}/* files
                        for path, ftype_size in archive_dict.find_by_prefix(
                                member + '/'):
                            logger.debug('find_by_prefix: %s [%s]',
                                         path, str(ftype_size))
                            file_type, size = ftype_size
                            member_set.add(path)
                            self.selected_num += 1
                            self.selected_size += size
                            found = True
                            if file_type == InfoDB.TYPE_DIR:
                                logger.debug('directory_set.add: %s', path)
                                directory_set.add(path)
                    if found:
                        member_check_dict[member] = True
                        is_target = True
                archive_dict.clear()  # re-use for next tar file
                if is_target:
                    logger.debug('gen_tarlist_dict[%d].append(%s)',
                                 gen, tar_path)
                    gen_tarlist_dict[gen].append(tar_path)
            else:
                # use all tar files
                gen_tarlist_dict[gen].append(tar_path)
            if self.progress_enabled:
                now = time.time()
                if now >= self.next_time:
                    self.next_time = now + self.progress_interval
                    self.progress_for_schedule(now)
        if self.progress_enabled:
            self.progress_for_schedule(time.time())
            self.progress_newline()
        if self.search_target:
            # update for progress
            self.total_num = self.selected_num
            self.total_size = self.selected_size

        for member, found in member_check_dict.items():
            logger.debug('check member_check_dict: %s, %s', member, found)
            if not found:
                raise GfptarError('The specified file is not found'
                                  ' in tar files: ' + member)
        member_check_dict.clear()
        archive_dict.clear()
        return gen_tarlist_dict, directory_set, member_set, error_num

    def extract_main(self):
        (gen_tarlist_dict, directory_set,
         member_set, error_num) = self.extract_schedule()

        # NOTE: Directories are created in threads
        # self.extract_directories(directory_set)  # slow
        # self.extract_directories_fast(directory_set)

        self.extracted_num = 0
        self.extracted_size = 0
        self.start_time = time.time()
        self.next_time = self.start_time + self.progress_interval
        if self.progress_enabled:
            self.progress_for_extract(self.start_time)

        self.gfsched_init()

        # sort by generation ID
        if self.MT_enabled():
            for gen, tarlist in sorted(gen_tarlist_dict.items()):
                self.extract_from_archives_MT(tarlist, member_set)
        else:
            for gen, tarlist in sorted(gen_tarlist_dict.items()):
                self.extract_from_archives(tarlist, member_set)
        if self.progress_enabled:
            self.progress_for_extract(time.time())
            self.progress_newline()

        self.update_stat_for_directories(directory_set)
        self.db.close()
        self.db.unlink()

        # use sync() instead of applying fsync() to each local file
        if self.outdir_is_local:
            logger.debug("sync() START: %f", time.time())
            os.sync()
            logger.debug("sync() END  : %f", time.time())

        if error_num > 0:
            raise GfptarError("Some .tar.* or .db.gz are broken:"
                              f" count={error_num}")

    def extract_directories(self, directory_set):
        logger.debug('extract_directories')
        # NOTE: slow on Gfarm
        for d in directory_set.iterator(sort='ASC'):
            if self.is_canceled():
                raise self.error_canceled()
            url_str = self.outdir_url.url_join(d)
            dir_url = GfURL.init(url_str, dry_run=self.dry_run)
            try:
                dir_url.mkdir()
            except MemoryError:
                raise
            except Exception:
                dir_url.makedirs()
            self.info('created(D): {}', dir_url.url_str)
            self.created_directory_set.add(dir_url.path)

    def extract_directories_fast(self, directory_set):
        logger.debug('extract_directories_fast: start')
        # faster implementation for gfmkdir
        created_set = self.created_directory_set
        for d in directory_set.iterator(sort='DESC'):
            if self.is_canceled():
                raise self.error_canceled()
            url_str = self.outdir_url.url_join(d)
            dir_url = GfURL.init(url_str, dry_run=self.dry_run)
            logger.debug('extract_directories_fast: '
                         f"out_url={self.outdir_url.url_str}, d={d}")
            # url.path is normalized
            if dir_url.path in created_set:
                logger.debug('skip (already created): %s', url_str)
                continue
            parent_url = dir_url.parent
            if parent_url.path in created_set:  # parent exists
                dir_url.mkdir()
                created_set.add(dir_url.path)
                self.info('created(D): {}', dir_url.url_str)
            else:  # no parent
                dir_url.makedirs()
                created_set.add(dir_url.path)
                self.info('created(D): {}', dir_url.url_str)
                created_set.add(parent_url.path)
                self.info('created(D): {}', parent_url.url_str)
                for p in parent_url.parent_iter:
                    path = p.path
                    if path == '.' or path == '/':
                        continue
                    created_set.add(path)
                    self.info('created(D): {}', p.url_str)

    def update_stat_for_directories(self, directory_set):
        logger.debug('update_stat_for_directories')
        self.clear_canceled()

        self.start_time = time.time()
        self.next_time = self.start_time + self.progress_interval
        total = len(directory_set)
        i = 0
        # DESC: process from leaves
        for d in directory_set.iterator(sort='DESC'):
            if self.is_canceled():
                logger.debug('Canceled (extract 3)')
                break
            tarinfo = self.dirstat_dict.get(d)
            if tarinfo is None:
                logger.warning('No information of the directory: %s', d)
                continue
            url_str = self.outdir_url.url_join(d)
            dir_url = GfURL.init(url_str, dry_run=self.dry_run)
            if self.same_owner:
                dir_url.chmod(tarinfo.mode, mtime=tarinfo.mtime,
                              user=tarinfo.uname, group=tarinfo.gname)
            else:
                dir_url.chmod(tarinfo.mode, mtime=tarinfo.mtime)
            self.info('update_stat: {}, mode={}', d, oct(tarinfo.mode))
            i += 1
            if self.progress_enabled:
                now = time.time()
                if now >= self.next_time:
                    self.next_time = now + self.progress_interval
                    self.progress_for_chmod(now, i, total)
        if self.progress_enabled:
            now = time.time()
            self.progress_for_chmod(now, i, total)
            self.progress_newline()

    def extract_from_archives(self, tarlist, member_set):
        self.lock_init(False)
        index = 0
        for target in tarlist:
            logger.debug('target tar: %s', target)
            index += 1
            self.extract_from_a_tar(index, target, member_set)

    def extract_from_archives_MT(self, tarlist, member_set):
        self.lock_init(True)
        with concurrent.futures.ThreadPoolExecutor(
                max_workers=self.max_workers) as executor:
            self.futures = {}  # tar filenames
            index = 0
            for target in tarlist:
                logger.debug('target tar: %s', target)
                index += 1
                t = executor.submit(self.extract_from_a_tar,
                                    index, target, member_set)
                self.futures[t] = target

            for t in concurrent.futures.as_completed(self.futures,
                                                     timeout=None):
                exc = t.exception()
                if exc:
                    logger.debug('%s: %s', self.futures[t],
                                 convert_message(exc))
                    self.print_trace(exc)
                    if not self.is_canceled():
                        self.cancel()
                        # DO NOT cancel threads because it will freeze.
                        # for t2 in self.futures:
                        #     if t == t2:
                        #         continue
                        #     logger.error('Canceled: %s', self.futures[t2])
                        #     t2.cancel()
                    raise exc

    def extract_from_a_tar(self, index, target, member_set):
        try:
            self.extract_from_a_tar0(index, target, member_set)
        except Exception as e:
            if self.is_canceled():
                self.cancel()
                raise self.error_canceled() from e
            else:
                raise

    def extract_from_a_tar0(self, index, target, member_set):
        logger.debug('extract_from_a_tar start: index=%d', index)
        if self.is_canceled():
            logger.debug('Canceled (extract 1): name=%s', target)
            return
        arch_url = GfURL.init(target,
                              use_gfarm_command=self.use_gfarm_command)
        tar = GfTarFile.extract_open(arch_url, self.bufsize)
        try:
            self.extract_from_a_tar1(index, target, member_set, tar)
            with self.lock():
                self.info('DONE: {}', arch_url.url_str)
        finally:
            tar.close()

    def extract_from_a_tar1(self, index, target, member_set, tar):
        with self.lock():
            members_num = len(member_set)
        is_local = self.outdir_url.is_local()
        while True:
            if self.is_canceled():
                logger.debug('Canceled (extract 2): name=%s', target)
                break
            try:
                tarinfo = tar.next()
            except MemoryError:
                raise
            except Exception as e:
                logger.debug('tar.next()', exc_info=e)
                logger.warning(f"{target}: SKIPPED: invalid or empty tar: "
                               f" {str(e)}")
                tarinfo = None
            if tarinfo is None:
                break
            if self.exclude_match(tarinfo.name):
                continue  # skip
            if members_num > 0:
                with self.lock():
                    # perfect match
                    if tarinfo.name not in member_set:
                        continue  # skip
                    # NOTE: DO NOT remove for --update
                    # member_set.remove(tarinfo.name)
            # members_num == 0 -> extract all

            # ex. //a/b/c// -> a/b/c
            outfile = tarinfo.name.strip('/')  # relative path only
            outurl_str = self.outdir_url.url_join(outfile)
            outurl = GfURL.init(outurl_str, dry_run=self.dry_run)

            if is_local:
                if tarinfo.isdir():
                    dir_url = outurl
                else:
                    dir_url = outurl.parent
                with ignore_exception(True):  # may be race condition
                    if not dir_url.exists():
                        dir_url.makedirs()  # default 0700
            else:  # speed up directory creation for Gfarm using SQLite3
                # create directories if not exist
                if tarinfo.isdir():
                    dir_url = outurl
                    dir_path = outurl.path
                else:
                    dir_url = outurl.parent
                    dir_path = dir_url.path
                with self.lock():
                    dir_exists = dir_path in self.created_directory_set
                if not dir_exists:
                    if not dir_url.exists():
                        with ignore_exception(True):  # may be race condition
                            dir_url.makedirs()  # default 0700
                        self.info('created(D): {}', dir_url.url_str)
                    with self.lock():
                        self.created_directory_set.add(dir_path)

            if tarinfo.isfile():
                target_host = self.select_a_target_host(outurl, index)
                index += 1
                user = None
                group = None
                if self.same_owner:
                    user = tarinfo.uname
                    group = tarinfo.gname
                inf = tar.extractfile(tarinfo)  # io.BufferedReader
                logger.debug('extract,file(before): %s, %s',
                             outurl_str, outfile)
                try:
                    size = outurl.copy_from(inf, self.bufsize,
                                            overwrite=True,
                                            mode=tarinfo.mode,
                                            mtime=tarinfo.mtime,
                                            user=user, group=group,
                                            use_fsync=self.use_fsync_for_x,
                                            hostname=target_host)
                finally:
                    inf.close()
                logger.debug('extract,file: %s, %d', outfile, size)
                with self.lock():
                    self.info('extracted(F): {}', outfile)
            elif tarinfo.isdir():
                # NOTE: already created
                logger.debug('extract,dir: %s', outfile)
                fattr = FileAttr1(tarinfo.mode, tarinfo.mtime,
                                  tarinfo.uname, tarinfo.gname)
                with self.lock():
                    self.dirstat_dict[outfile] = fattr
            elif tarinfo.issym():
                logger.debug('extract,link: %s, %s', outfile,
                             tarinfo.linkname)
                try:
                    outurl.symlink(tarinfo.linkname)
                except Exception:
                    if outurl.exists():
                        outurl.remove()
                        outurl.symlink(tarinfo.linkname)
                    else:
                        raise
                if self.same_owner:
                    outurl.chmod(tarinfo.mode, mtime=tarinfo.mtime,
                                 user=tarinfo.uname, group=tarinfo.gname,
                                 follow_symlinks=False)
                else:
                    outurl.chmod(tarinfo.mode, mtime=tarinfo.mtime,
                                 follow_symlinks=False)
                with self.lock():
                    self.info('extracted(S): {}', outfile)
            else:
                logger.warning('unsupported type: %s: %s',
                               outfile, tarinfo.type)
            if self.progress_enabled:
                with self.lock():
                    self.extracted_num += 1
                    if tarinfo.isfile():
                        self.extracted_size += tarinfo.size
                    now = time.time()
                    if now >= self.next_time:
                        self.next_time = now + self.progress_interval
                        self.progress_for_extract(now)

    def _humanize_si(self, n):
        return humanize_number(n)

    def _humanize_bin(self, n):
        return humanize_number(n, binary_prefix=True)

    def _humanize_raw(self, n):
        return int(n)

    def progress_print(self, msg):
        if self.debug:
            logger.debug('Progress: ' + msg)
        elif self.verbose:
            self.info('Progress: ' + msg)
        else:
            sys.stdout.write("\r")
            sys.stdout.write(msg)
            sys.stdout.write("  ")
            sys.stdout.flush()

    def progress_newline(self):
        if self.debug:
            pass
        elif self.verbose:
            pass
        elif self.quiet:
            pass
        else:
            sys.stdout.write("\n")

    def progress_for_schedule(self, now):
        sec = now - self.start_time
        sec_str = format_seconds(sec, minhour=True)
        total_num_str = self._humanize(self.total_num)
        total_size_str = self._humanize(self.total_size)
        if sec > 0:
            ent_per_sec = self.total_num / sec
        else:
            ent_per_sec = 0
        ent_per_sec_str = self._humanize(ent_per_sec)
        name = "schedule"
        if self.selected_num > 0:
            selected_num_str = self._humanize(self.selected_num)
            selected_size_str = self._humanize(self.selected_size)
            msg = (f"{name}: "
                   f"{selected_size_str}/{total_size_str}B "
                   f"{selected_num_str}/{total_num_str}Ent "
                   f"{sec_str} "
                   f"{ent_per_sec_str}Ent/s")
        else:
            msg = (f"{name}: "
                   f"{total_size_str}B "
                   f"{total_num_str}Ent "
                   f"{sec_str} "
                   f"{ent_per_sec_str}Ent/s")
        self.progress_print(msg)

    def progress_for_infodb(self, now):
        sec = now - self.start_time
        sec_str = format_seconds(sec, minhour=True)
        current_tar_num_str = self._humanize(self.current_tar_num)
        total_tar_num_str = self._humanize(self.total_tar_num)
        if self.current_ent_num > 0:
            current_ent_num_str = self._humanize(self.current_ent_num)
        else:
            current_ent_num_str = '?'
        if self.current_size > 0:
            current_size_str = self._humanize(self.current_size)
        else:
            current_size_str = '?'
        if sec > 0:
            ent_per_sec = self.current_ent_num / sec
        else:
            ent_per_sec = 0
        ent_per_sec_str = self._humanize(ent_per_sec)
        if self.gen_db:
            name = 'generate-db'
            msg = (f"{name}: "
                   f"{current_tar_num_str}/{total_tar_num_str}(DB) "
                   f"{current_size_str}B "
                   f"{current_ent_num_str}Ent "
                   f"{sec_str} "
                   f"{ent_per_sec_str}Ent/s")
        else:
            name = 'load-db'
            msg = (f"{name}: "
                   f"{current_tar_num_str}/{total_tar_num_str}(DB) "
                   f"{sec_str} ")
        self.progress_print(msg)

    # lock required
    def progress_for_create(self, now):
        sec = now - self.start_time
        sec_str = format_seconds(sec, minhour=True)
        if self.listing:
            percent_str = '?'
        else:
            if self.selected_num > 0:
                percent1 = self.stored_num * 100 / self.selected_num
            else:
                percent1 = 0
            if self.selected_size > 0:
                percent2 = self.stored_size * 100 / self.selected_size
                percent = (percent1 + percent2) / 2
            else:
                percent = percent1
            percent_str = f"{percent:.0f}"
        if sec > 0:
            bytes_per_sec = self.stored_size / sec
            ent_per_sec = self.stored_num / sec
        else:
            bytes_per_sec = 0
            ent_per_sec = 0
        stored_num_str = self._humanize(self.stored_num)
        sel_num_str = self._humanize(self.selected_num)
        total_num_str = self._humanize(self.total_num)
        stored_size_str = self._humanize(self.stored_size)
        sel_size_str = self._humanize(self.selected_size)
        # total_size_str = self._humanize(self.total_size)
        bytes_per_sec_str = self._humanize(bytes_per_sec)
        ent_per_sec_str = self._humanize(ent_per_sec)
        msg = (f"{self.cmd_name}: {percent_str}% "
               f"{stored_size_str}/{sel_size_str}B "
               f"{stored_num_str}/{sel_num_str}/{total_num_str}Ent "
               f"{sec_str} "
               f"{bytes_per_sec_str}B/s "
               f"{ent_per_sec_str}Ent/s")
        self.progress_print(msg)

    # lock required
    def progress_for_extract(self, now):
        sec = now - self.start_time
        sec_str = format_seconds(sec, minhour=True)
        if self.total_num > 0:
            percent1 = self.extracted_num * 100 / self.total_num
        else:
            percent1 = 0
        if self.total_size > 0:
            percent2 = self.extracted_size * 100 / self.total_size
            percent = (percent1 + percent2) / 2
        else:
            percent = percent1
        percent_str = f"{percent:.0f}"
        if sec > 0:
            bytes_per_sec = self.extracted_size / sec
            ent_per_sec = self.extracted_num / sec
        else:
            bytes_per_sec = 0
            ent_per_sec = 0
        extracted_num_str = self._humanize(self.extracted_num)
        total_num_str = self._humanize(self.total_num)
        total_size_str = self._humanize(self.total_size)
        extracted_size_str = self._humanize(self.extracted_size)
        bytes_per_sec_str = self._humanize(bytes_per_sec)
        ent_per_sec_str = self._humanize(ent_per_sec)
        name = "extract"
        msg = (f"{name}: {percent_str}% "
               f"{extracted_size_str}/{total_size_str}B "
               f"{extracted_num_str}/{total_num_str}Ent "
               f"{sec_str} "
               f"{bytes_per_sec_str}B/s "
               f"{ent_per_sec_str}Ent/s")
        self.progress_print(msg)

    def progress_for_chmod(self, now, current, total):
        sec = now - self.start_time
        sec_str = format_seconds(sec, minhour=True)
        if total > 0:
            percent = current * 100 / total
        else:
            percent = 0
        percent_str = f"{percent:.0f}"
        if sec > 0:
            ent_per_sec = current / sec
        else:
            ent_per_sec = 0
        current_str = self._humanize(current)
        total_str = self._humanize(total)
        ent_per_sec_str = self._humanize(ent_per_sec)
        msg = (f"chmod: {percent_str}% "
               f"{current_str}/{total_str}Ent "
               f"{sec_str} "
               f"{ent_per_sec_str}Ent/s")
        self.progress_print(msg)

    def cmd_list_simple(self, indir):
        return self.cmd_list(indir, verbose=False)

    def cmd_list_verbose(self, indir):
        return self.cmd_list(indir, verbose=True)

    def cmd_list(self, indir, verbose=True):
        if self.test_mode:
            outf = open(os.devnull, 'w')
        else:
            outf = sys.stdout
        try:
            return self.cmd_list0(outf, indir, verbose)
        finally:
            if self.test_mode:
                outf.close()

    def cmd_list0(self, outf, indir, verbose):
        self.options_init()
        quiet = self.quiet
        indir_url = GfURL.init(indir)

        def print_quiet(serial, gen, path, fattr):
            pass

        def print_verbose(serial, gen, path, fattr):
            name = path
            if fattr.ftype == InfoDB.TYPE_DIR:
                name = name + '/'
            elif fattr.ftype == InfoDB.TYPE_SYMLINK:
                name = name + ' -> ' + fattr.linkname
            dt_object = datetime.datetime.fromtimestamp(fattr.mtime)
            mtime_str = dt_object.strftime('%Y-%m-%d %H:%M')
            outf.write('g%d_%04d %s %04o %8s/%-8s %9d %s %s\n' % (
                gen, serial, fattr.ftype, fattr.mode, fattr.user,
                fattr.group, fattr.size, mtime_str, name))

        def print_simple(serial, gen, path, fattr):
            if fattr.ftype == InfoDB.TYPE_DIR:
                print(f"{fattr.ftype} {path}/", file=outf)
            else:
                print(f"{fattr.ftype} {path}", file=outf)

        if quiet:
            print_func = print_quiet
            resolve_ugmap = False
        elif verbose:
            print_func = print_verbose
            resolve_ugmap = True
        else:
            print_func = print_simple
            resolve_ugmap = False

        count = 0
        for serial, gen, ent_path, fattr in self.list_entries(
                indir_url, resolve_ugmap=resolve_ugmap):
            if self.is_canceled():
                break
            print_func(serial, gen, ent_path, fattr)
            count += 1
        self.tmpdir.cleanup()
        return count

    def list_entries(self, indir_url, resolve_ugmap=False, progress=False):
        # SEE ALSO: extract_schedule()
        error_num = 0
        for serial, gen, tar_path, dbgz_path, db_path in \
                self.list_tar_dbgz_db(indir_url, sort=True, progress=progress):
            if db_path is None:
                error_num += 1
                continue
            if tar_path is None:
                error_num += 1
                # FALLTHROUGH
            for ent_path, fattr in InfoDB.list_entries_from_db(
                    db_path, resolve_ugmap=resolve_ugmap):
                yield serial, gen, ent_path, fattr
            logger.debug('os.remove: %s', db_path)
            os.remove(db_path)
        if error_num > 0:
            raise GfptarError("Some .tar.* or .db.gz are broken:"
                              f" count={error_num}")

    def sort_tar_or_db_list(self, tar_or_db_list):
        def id_key(value):
            return value[0]  # 0:serial, 1:gen, 2:path

        tar_or_db_list.sort(key=id_key)
        return tar_or_db_list

    def list_tar_list_dbgz_from_dir(self, indir_url):
        PATT_TAR = r'^.*\.tar(\.\w{1,5})?$'  # *.tar or *.tar.*
        PATT_TAR_GEN1 = r'^(\d+)_.+'
        PATT_TAR_GEN_ALL = r'^g(\d+)_(\d+)_.+'
        tar_pattern = re.compile(PATT_TAR)
        tar_gen1_pattern = re.compile(PATT_TAR_GEN1)
        tar_gen_all_pattern = re.compile(PATT_TAR_GEN_ALL)
        dbgz_pattern = re.compile(InfoDB.PATT_DBGZ)
        tar_list = []
        dbgz_list = []
        for ent in indir_url.listdir(recursive=False):
            path = ent.path
            base = os.path.basename(path)
            db_match = dbgz_pattern.match(base)
            if db_match:
                gen = db_match.group(1)
                serial = db_match.group(2)
                # int("0001") -> 1
                dbgz_list.append((int(serial), int(gen), path))
                continue
            if ent.path.endswith(self.LIST_SUFFIX):  # ignore *.lst
                self.progress_newline()
                logger.warning(f"NOTICE: {ent.path}:"
                               " This file (*.lst) is no longer needed."
                               " Please remove it.")
                continue
            if not tar_pattern.match(base):  # ignore not *.tar.*
                continue
            t1_match = tar_gen1_pattern.match(base)
            if t1_match:
                # generation number = 1
                gen = '1'
                # ex. 0001 (str)
                serial = t1_match.group(1)
            else:
                t_match = tar_gen_all_pattern.match(base)
                if t_match:
                    gen = t_match.group(1)
                    # ex. 0001 (str)
                    serial = t_match.group(2)
                else:
                    # ignore irrelevant file
                    continue
            # int("0001") -> 1
            tar_list.append((int(serial), int(gen), path))
        return tar_list, dbgz_list

    # yield tuple(serial, gen, tar_path, dbgz_path, db_path)
    def list_tar_dbgz_db(self, indir_url, sort=False, progress=False):
        logger.debug('list_tar_dbgz_db: start: %s', indir_url.url_str)
        try:
            yield from self.list_tar_dbgz_db0(indir_url,
                                              sort=sort, progress=progress)
        finally:
            logger.debug('list_tar_dbgz_db: done: %s', indir_url.url_str)

    def list_tar_dbgz_db0(self, indir_url, sort=False, progress=False):
        tar_list, dbgz_list = self.list_tar_list_dbgz_from_dir(indir_url)

        tar_dict = {}
        dbgz_dict = {}
        for serial, gen, tar_path in tar_list:
            tar_dict[serial] = (gen, tar_path)

        for serial, gen, dbgz_path in dbgz_list:
            dbgz_dict[serial] = (gen, dbgz_path)
            gen_tar = tar_dict.get(serial, None)
            if gen_tar is None:
                logger.error(f"lost *.tar.* tied to {dbgz_path}")
                tar_dict[serial] = (gen, None)

        for serial, gen, tar_path in tar_list:
            gen_dbgz = dbgz_dict.get(serial, None)
            if gen_dbgz is None:
                dbgz_fname = InfoDB.dbgz_filename(gen, serial)
                logger.info(f"lost {dbgz_fname} (auto re-creation)"
                            f" tied to {tar_path}")
                dbgz_dict[serial] = (gen, None)

        # for progress
        self.total_tar_num = 0
        self.current_tar_num = 0

        tar_dbgz_list = []
        for serial, gen_dbgz in dbgz_dict.items():
            gen, dbgz_path = gen_dbgz
            gen_tar = tar_dict.get(serial, None)
            if gen_tar is not None:
                gen2, tar_path = gen_tar
            else:
                tar_path = None
            tar_dbgz_list.append((serial, gen, tar_path, dbgz_path))
            self.total_tar_num += 1
        if sort:  # sort by serial
            tar_dbgz_list.sort(key=lambda x: x[0])

        del dbgz_dict
        del tar_dict
        del tar_list
        del dbgz_list

        self.start_time = time.time()
        self.save_e = None
        cancel = False

        sigs = [signal.SIGHUP, signal.SIGINT, signal.SIGTERM]
        orig_sig_handler = {}
        for sig in sigs:
            orig_sig_handler[sig] = signal.getsignal(sig)

        self.sig_ignore()

        # Concurrent execution for each generation
        with multiprocessing.Manager() as manager:  # subprocess
            lock = manager.Lock()
            share_cancel = manager.Value('i', 0)

            def sig_handler(signum, frame):
                if share_cancel.value == 0:
                    sys.stderr.write(f"\nInterrupt (signal={signum})\n")
                    share_cancel.value = 1
                    self.cancel()

            signal.signal(signal.SIGHUP, sig_handler)
            signal.signal(signal.SIGINT, sig_handler)
            signal.signal(signal.SIGTERM, sig_handler)
            signal.signal(signal.SIGPIPE, signal.SIG_IGN)

            self.current_ent_num = 0
            self.current_size = 0
            if progress:
                now = time.time()
                self.progress_for_infodb(now)

            def result_one(one):
                future, gen, serial, tar_path, dbgz_path = one
                try:
                    db_path, dbgz_path, ent_num, size = future.result()
                    self.current_ent_num += ent_num
                    self.current_size += size
                    self.current_tar_num += 1
                    if progress:
                        now = time.time()
                        self.progress_for_infodb(now)
                    yield serial, gen, tar_path, dbgz_path, db_path
                except Exception as e:
                    # logger.error(f"{e}")
                    self.print_trace(e)
                    share_cancel.value = 1
                    if self.save_e is None:
                        self.save_e = e

            def result_all(previous_list):
                for one in previous_list:
                    yield from result_one(one)

            # concurrent processes per generation (gen)
            job_list = []
            with concurrent.futures.ProcessPoolExecutor(
                    max_workers=self.max_workers) as executor:
                for serial, gen, tar_path, dbgz_path in tar_dbgz_list:
                    if share_cancel.value != 0:
                        cancel = True
                        break
                    if len(job_list) >= self.max_workers:
                        # wait for a previous job
                        yield from result_one(job_list[0])
                        job_list = job_list[1:]  # remove first
                    if tar_path is None:
                        # missing tar file
                        db_path = None
                        # wait for all previous jobs
                        yield from result_all(job_list)
                        job_list = []
                        yield serial, gen, tar_path, dbgz_path, db_path
                        continue
                    if dbgz_path is None:
                        gen_db = True
                    else:
                        gen_db = self.gen_db
                    arg = (lock, share_cancel, self.tmpdir.name,
                           gen_db, gen, serial, tar_path,
                           self.bufsize, self.use_fsync, self.dry_run)
                    # InfoDB.generate_db_and_dbgz is classmethod,
                    # because ProcessPoolExecutor cannot serialize
                    # members of "GfptarProgram (self)" object.
                    future = executor.submit(InfoDB.generate_db_and_dbgz, arg)
                    job_list.append((future, gen, serial, tar_path, dbgz_path))
                yield from result_all(job_list)
        # no longer be able to access Manager
        for sig in sigs:
            # restore
            signal.signal(sig, orig_sig_handler[sig])

        if progress:
            now = time.time()
            self.progress_for_infodb(now)
            self.progress_newline()
        if self.save_e:
            raise self.save_e
        if cancel:
            raise self.error_canceled()


class InfoDB:
    TYPE_FILE = 'F'
    TYPE_DIR = 'D'
    TYPE_SYMLINK = 'S'
    TYPE_OTHER = '?'
    TABLE_ENTRY = 'path_entry'
    TABLE_USER = 'user_map'
    TABLE_GROUP = 'group_map'

    def __init__(self, db_path, clear=False):
        self.db = DB(db_path)
        self.fattr_dict = DBDict(self.db, FileAttr2, self.TABLE_ENTRY,
                                 clear=clear)
        self.user_dict = {}   # use memory for speed
        self.group_dict = {}  # use memory for speed

    # SEE ALSO: dbgz_filename()
    PATT_DBGZ = r'^g(\d+)_(\d+)_gfptar.db.gz$'

    @classmethod
    def db_filename(cls, gen, serial):
        # SEE ALSO: PATT_DBGZ
        # gen (>= 1): the generation number
        # serial  (>= 1): the serial number
        # ex. g2_0099_gfptar.db.gz
        return f"g{gen}_{int(serial):04}_gfptar.db"

    @classmethod
    def dbgz_filename(cls, gen, serial):
        return cls.db_filename(gen, serial) + '.gz'

    def add(self, tarinfo):
        t = tarinfo
        if t.isfile():
            ftype = self.TYPE_FILE
        elif t.isdir():
            ftype = self.TYPE_DIR
        elif t.issym():
            ftype = self.TYPE_SYMLINK
        else:
            ftype = self.TYPE_OTHER
        user_id = self.user_dict.get(t.uname, None)  # not uid
        if user_id is None:
            user_id = str(len(self.user_dict))
            self.user_dict[t.uname] = user_id
        group_id = self.group_dict.get(t.gname, None)  # not gid
        if group_id is None:
            group_id = str(len(self.group_dict))
            self.group_dict[t.gname] = group_id
        fattr = FileAttr2(t.mode, int(t.mtime), user_id, group_id,
                          t.size, t.linkname, ftype)
        self.fattr_dict[t.name] = fattr

    def commit_close(self):
        # unique id -> name
        rev_user_dict = DBDict(self.db, StrObj, self.TABLE_USER)
        rev_group_dict = DBDict(self.db, StrObj, self.TABLE_GROUP)
        for k, v in self.user_dict.items():
            rev_user_dict[v] = k
        for k, v in self.group_dict.items():
            rev_group_dict[v] = k
        self.db.commit()
        self.db.close()

    @staticmethod
    def compress_copy(in_db_path, out_dbgz_path, bufsize, use_fsync,
                      dry_run, move=True):
        dbgz_path_tmp = out_dbgz_path + '.tmp'
        dbgz_url_tmp = GfURL.init(dbgz_path_tmp, dry_run=dry_run)
        db_url = GfURL.init(in_db_path)

        def remove(gfurl):
            if gfurl.exists():
                gfurl.remove()

        remove(dbgz_url_tmp)
        try:
            with dbgz_url_tmp.writeopen(use_fsync=use_fsync) as outf:
                proc = Compress.compress(Compress.gzip_prog, outf)
                with db_url.readopen() as inf:
                    shutil.copyfileobj(inf, proc.stdin, bufsize)
                proc.stdin.close()
                ret = proc.wait()
                if ret != 0:
                    raise GfException('{}: returncode={}'.format(
                        ' '.join(proc.args), ret))
            # atomic operation to avoid leaving junk files
            dbgz_url_tmp.rename(out_dbgz_path)
        except Exception:
            remove(dbgz_url_tmp)
            raise

        logger.debug('created(.db.gz): %s', out_dbgz_path)
        if move:
            db_url.remove()

    @staticmethod
    def decompress_copy(in_dbgz_path, out_db_path, bufsize, use_fsync):
        # out_db_path is stored in tempfile.TemporaryDirectory
        dbgz_url = GfURL.init(in_dbgz_path)
        db_url = GfURL.init(out_db_path)
        if db_url.exists():
            db_url.remove()
        with dbgz_url.readopen() as inf:
            proc = Compress.decompress(Compress.gzip_prog, inf)
            with db_url.writeopen(use_fsync=use_fsync) as outf:
                shutil.copyfileobj(proc.stdout, outf, bufsize)
            proc.stdout.close()
            ret = proc.wait()
            if ret != 0:
                raise GfException('{} (from {} to {}): returncode={}'.format(
                    ' '.join(proc.args), in_dbgz_path, out_db_path, ret))
        return db_url

    signal_initialized = False
    signal_canceled = False

    @classmethod
    def signal_init(cls):
        if not cls.signal_initialized:
            def sig_handler(signum, frame):
                if not cls.signal_canceled:
                    pid = os.getpid()
                    logger.info(f"Interrupt (signal={signum}) (PID={pid})")
                    cls.signal_canceled = True

            signal.signal(signal.SIGHUP, sig_handler)
            signal.signal(signal.SIGINT, sig_handler)
            signal.signal(signal.SIGTERM, sig_handler)
            signal.signal(signal.SIGPIPE, signal.SIG_IGN)
            cls.signal_initialized = True
            logger.debug('InfoDB.signal_init')

    # return db_path, dbgz_path, ent_num, size
    @classmethod
    def generate_db_and_dbgz(cls, args):
        (lock, share_cancel, db_dir, gen_db, gen, serial,
         tar_path, bufsize, use_fsync, dry_run) = args
        if share_cancel.value != 0:
            # logger.debug('Canceled (2)')
            return None, None, 0, 0
        InfoDB.signal_init()

        tar_url = GfURL.init(tar_path)
        tardir_url = tar_url.parent
        dbgz_fname = InfoDB.dbgz_filename(gen, serial)
        dbgz_path = tardir_url.url_join(dbgz_fname)
        dbgz_url = GfURL.init(dbgz_path)

        # db_dir is local filesystem
        db_path = os.path.join(db_dir, cls.db_filename(gen, serial))

        num = 0
        size = 0
        if not gen_db and dbgz_url.exists():
            logger.debug('not generate, decompress dbgz: %s', dbgz_path)
            use_fsync2 = False  # disable fsync for workdir
            InfoDB.decompress_copy(dbgz_path, db_path, bufsize, use_fsync2)
            # NOTE: slow
            # infodb = InfoDB(db_path)
            # num = len(infodb.fattr_dict)
            # infodb.db.close()
        else:
            # generate db and dbgz
            tar = GfTarFile.extract_open(tar_url, bufsize)
            infodb = InfoDB(db_path, clear=True)
            db_close = False
            has_error = True
            try:
                interval = 1  # sec. (for interrupt)
                next_check = time.time() + interval
                while True:
                    if cls.signal_canceled:
                        logger.info('Canceled')
                        break
                    now = time.time()
                    if now >= next_check:
                        # access manager.Value(): very high cost
                        if share_cancel.value != 0:
                            logger.info('Canceled')
                            has_error = True
                            break
                        next_check = now + interval
                    try:
                        t = tar.next()
                    except MemoryError:
                        raise
                    except Exception as e:
                        logger.warning(f"{tar_path}: SKIPPED:"
                                       f" invalid or empty tar: {str(e)}")
                        t = None
                    if t is None:
                        break
                    infodb.add(t)
                    num += 1
                    size += t.size
                # success

                infodb.commit_close()
                db_close = True
                InfoDB.compress_copy(db_path, dbgz_path, bufsize,
                                     use_fsync, dry_run, move=False)
            except Exception:
                has_error = True
                raise
            finally:
                with ignore_exception(has_error):
                    tar.close()
                with ignore_exception(has_error):
                    if not db_close:
                        infodb.db.close()
        return db_path, dbgz_path, num, size

    @staticmethod
    def list_entries_from_db(db_path, resolve_ugmap=False, sort='ASC'):
        db = DB(db_path)
        has_error = False
        try:
            fattr_dict = DBDict(db, FileAttr2, InfoDB.TABLE_ENTRY)
            if resolve_ugmap:
                user_dict = DBDict(db, StrObj, InfoDB.TABLE_USER)
                group_dict = DBDict(db, StrObj, InfoDB.TABLE_GROUP)
                user_dict_mem = {}
                group_dict_mem = {}
                # cache in memory
                for k, v in user_dict.items():
                    user_dict_mem[k] = v
                for k, v in group_dict.items():
                    group_dict_mem[k] = v
            for path, fattr in fattr_dict.iterator(sort=sort):
                if resolve_ugmap:
                    # unique id -> name
                    fattr.user = user_dict_mem.get(fattr.user, '???')
                    fattr.group = group_dict_mem.get(fattr.group, '???')
                yield path, fattr
        except Exception:
            has_error = True
            raise
        finally:
            with ignore_exception(has_error):
                db.close()


progname = os.path.basename(__file__)


__doc__ = """
gfptar - archive files in parallel

Example of --create (Local to Gfarm):
  Command line:
    gfptar -c gfarm:/home/gfuser1/out1 -C /home/user1 ./dir1

  Input files (in /home/user1):
    dir1/test0000.data
    ...
    dir1/test9999.data

  Output files in archive directory:
    gfarm:/home/gfuser1/out1/0001_dir_test0000.data..dir_test0999.data.tar.gz
    gfarm:/home/gfuser1/out1/g1_0001_gfptar.db.gz
    ...
    gfarm:/home/gfuser1/out1/0010_dir_test9000.data..dir_test9999.data.tar.gz
    gfarm:/home/gfuser1/out1/g1_0010_gfptar.db.gz

Example of --extract (Gfarm to Local):
  Command line:
    gfptar -x /home/user1/dir2 gfarm:/home/gfuser1/out1

  Input archive directory:
    gfarm:/home/gfuser1/out1

  Output files:
    /home/user1/dir2/dir1/test0000.data
    ...
    /home/user1/dir2/dir1/test9999.data

Contents of gMM_NN_gfptar.db.gz file (SQLite3 and gzip):
  MM: the generation number for each append operation
  NN: the serial number
  table 'path_entry': map of path name to JSON string (BLOB)
    json.dumps([ file_mode (int), mtime (int),
                 user_unique_id (int), group_unique_id (int),
                 size (int), symlink_path, file_type (D,F,S) ]
    file_type 'D': directory
    file_type 'F': file
    file_type 'S': symbolic link
  table 'user_map' : map of unique id (not uid) to user name
  table 'group_map': map of unique id (not gid) to group name

SI prefix or Binary prefix:
  SI prefix:
    - 1k  = 10^3 = 1000^1 (kilo) (not K)
    - 1M  = 10^6 = 1000^2 (mega)
    - ...

  Binary prefix:
    - 1Ki = 2^10 = 1024^1 (kibi)
    - 1Mi = 2^20 = 1024^2 (mebi)
    - ...

Limitations:
  - Hard links are not preserved.
  - File names cannot include newline characters.
  - Subsecond (less than a second) for mtime is not preserved.

Envrionment variables:
  - GFPTAR_USE_ORJSON: Use orjson instead of json.
    (1 ... enable, 0 ... disable) (default: 1)
  - GFMKDIR_PLUS
  - GFCHMOD_PLUS
  - GFREG_PLUS

Options:
  -c, --create=OUTDIR       Create tar files in OUTDIR from MEMBERs.
  -r, --append=OUTDIR       Append files to new tar files.
  -u, --update=OUTDIR       Append only files (to new tar files) newer than
                            same entries in existing tar files.
                            This option can also be used to resume operation
                            after interruption (including errors).
                            (Sufficient space for --workdir is required)
                            (About 300MiB is used per existing 1000000 entries)
  -x, --extract=OUTDIR      Extract all entries or specified MEMBERs
                            from INDIR to OUTDIR.
  -t, --list=DIR            List the members of DIR.
                            (--verbose to see more details)
  -C, --basedir=DIR         Change to directory for MEMBERs [default: .]
  --exclude=PATTERN         Exclude files matching wildcard patterns.
                            (ex. --exclude=*.txt)
                            (ex. --exclude=*/dirname/*)
                            (ex. --exclude=*/??_abcde.pdf)
                            (https://docs.python.org/ja/3/library/fnmatch.html)
  -X, --exclude-from=FILE   Exclude files matching wildcard patterns listed
                            in FILE.
  --workdir=DIR             Local directory for temporary files.
                            (default: {tempdir})
  -j, --jobs=NUM            The number of jobs to copy per tar file
                            in parallel.
                            [default: 4]
  -s, --size=BYTES          Maximum assumed size per output file.
                            [default: 200Mi]
  --ratio=RATIO             Assumed compression ratio (%). [default: 50]
  --max-entries-per-tar=NUM
                            The maximum number of entries per tar file.
                            Too many entries in a tar may use a lot of memory,
                            and may work slowly.
                            [default: 100k]
  -T, --type=TYPE           Compression type and tar file suffix.
                                gz : use gzip  (*.tar.gz)
                                bz2: use bzip2 (*.tar.bz2)
                                xz : use xz    (*.tar.xz)
                                no : no compression (*.tar)
                                (other): --use-compress-program is required
                            (default: gz)
  -I, --use-compress-program=COMMAND
                            Command to compress or decompress tar files
                            when using --type other than gz, bz2 or xz.
                            (The command must accept -d option to decompress)
                            (ex. lz4, lzip, lzop)
  --gzip-program=COMMAND    gzip command  (ex. pigz)   [default: gzip]
  --bzip2-program=COMMAND   bzip2 command (ex. pbzip2) [default: bzip2]
  --xz-program=COMMAND      xz command (ex. environment XZ_OPT="-T4" for speed)
                            [default: xz]
  --generate-db             Regenerate gfptar*_info.db.gz
  --same-owner              Extract files with the same ownership.
                            (for euid=0 on local, or gfarmroot on Gfarm)
  --disable-gfarm-command   Disable the use of gfreg and gfexport
                            for tar files on gfarm2fs.
  --disable-fsync           Disable calling fsync() before close().
  --gfsched-interval=SEC    Interval of updating candidate hosts to write.
                            (for Gfarm URL only) [default: 120]
  --progress-interval=SEC   Interval of updating progress. [default: 1.0]
  --encoding=CODEC          Codec for filename encoding.
             (https://docs.python.org/3/library/codecs.html#standard-encodings)
                            [default: utf-8]
  --bufsize=BYTES           Buffer size to copy. [default: 1Mi]
  --disable-tempdb-aggregation
                            This option only effective with --update option.
                            When there are few files in MEMBERs, this option
                            may improve the execution speed.
                            However, if more than about {max_openfiles}
                            tar files in archive directory exist,
                            the execution with this option will fail.
  --max-entries-for-tempdb-aggregation=NUM
                            The maximum number of entries per working DB file.
                            This option only effective with --update option.
                            Normally, you don't need to change this value.
                            If a smaller value is specified, performance
                            may degrade.
                            (0: unlimited (using one file as working DB))
                            [default: 0]
  --progress-unit=TYPE      Unit type for progress
                                si : SI prefix
                                bin: Binary prefix
                                raw: no conversion
                            [default: si]
  --memory=BYTES            Upper limit of memory size (bytes)
                            (default: no limit) (ex. 2Gi)
  --test                    Run short tests (-q option is recommended)
  --test-long               Run long  tests (-q option is recommended)
  --test-workdir-local=DIR  Local directory for test [default: /tmp]
  --test-workdir-gfarm=DIR  Gfarm directory for test [default: gfarm:/tmp]
                            (A path in gfarm2fs can be specified)
  --dummy-num=NUM           The number of dummy input entries for --create.
                            (MEMBER arguments are ignored.)
                            (1000 files are created per directory)
                            (default: disabled) (ex. 1M)
  --dummy-size-min=BYTES    Minimum size of dummy files [default: 0]
  --dummy-size-max=BYTES    Maximum size of dummy files [default: 0]
  --debug-sleep=SEC         Sleep time per file for debug [default: 0.0]
  --dry-run                 Not create output files.
  -q, --quiet               Quiet messages.
  -v, --verbose             Verbose output.
  -d, --debug               Debug mode.
  -?, --help                Show this help and exit.

Usage:
  {f} [options] [--exclude=PATTERN]... -c OUTDIR [-C DIR] [--] MEMBER...
  {f} [options] [--exclude=PATTERN]... -r OUTDIR [-C DIR] [--] MEMBER...
  {f} [options] [--exclude=PATTERN]... -u OUTDIR [-C DIR] [--] MEMBER...
  {f} [options] [--exclude=PATTERN]... -x OUTDIR [--] INDIR [MEMBER...]
  {f} [options] -t DIR
  {f} [options] --test
  {f} [options] --test -C DIR MEMBER...
  {f} [options] --test-long
  {f} -? | --help
""".format(f=progname,
           tempdir=tempfile.gettempdir(),
           max_openfiles=get_max_openfiles())


_schema = Schema({
    '--list': Or(str, None),
    '--extract': Or(str, None),
    '--create': Or(str, None),
    '--append': Or(str, None),
    '--update': Or(str, None),
    '--basedir': Or(str, None),
    '--exclude': Or([str], None),
    '--exclude-from': Or(str, None),
    '--encoding': str,
    '--size': Use(unhumanize_number),
    '--bufsize': Use(unhumanize_number),
    '--type': Or(str, None),
    '--ratio': Use(int),
    '--jobs': Use(int),
    '--use-compress-program': Or(str, None),
    '--gzip-program': Or(str, None),
    '--bzip2-program': Or(str, None),
    '--xz-program': Or(str, None),
    '--disable-gfarm-command': bool,
    '--disable-fsync': bool,
    '--gfsched-interval': Use(int),
    '--disable-tempdb-aggregation': bool,
    '--max-entries-for-tempdb-aggregation': Use(unhumanize_number),
    '--generate-db': bool,
    '--same-owner': bool,
    '--workdir': Or(str, None),
    '--max-entries-per-tar': Use(unhumanize_number),
    '--progress-interval': Use(float),
    '--progress-unit': str,
    '--memory': Or(Use(unhumanize_number), None),
    '--test': bool,
    '--test-long': bool,
    '--test-workdir-local': Or(str, None),
    '--test-workdir-gfarm': Or(str, None),
    '--dummy-num': Or(Use(unhumanize_number), None),
    '--dummy-size-min': Use(unhumanize_number),
    '--dummy-size-max': Use(unhumanize_number),
    '--debug-sleep': Use(float),
    '--dry-run': bool,
    '--quiet': bool,
    '--verbose': bool,
    '--debug': bool,
    '--help': bool,
    '--': bool,
    'INDIR': Or(str, None),
    'MEMBER': [str],
})


if __name__ == '__main__':
    # Compatibility fix for Python 3.14+:
    # Force the use of 'fork' to avoid 'TypeError: cannot pickle _thread.lock'
    # which occurs when using 'spawn' or 'forkserver' with complex objects.
    # Ref: https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods  # noqa: E501
    multiprocessing.set_start_method('fork', force=True)

    gfptar = GfptarProgram(progname)
    gfptar.run()
