def worker_dsh_cp(addr):
    '''do remote copy to node'''

    nodename = NODESET.get_nodename_from_address(addr)
    if nodename == synctool.param.NODENAME:
        # do not copy to local node; files are already here
        return

    # the fileset already has been added to DSH_CP_CMD_ARR

    # create local copy of DSH_CP_CMD_ARR
    # or parallelism may screw things up
    dsh_cp_cmd_arr = DSH_CP_CMD_ARR[:]
    dsh_cp_cmd_arr.append('%s:%s' % (addr, DESTDIR))

    msg = 'copy %s to %s' % (FILES_STR, DESTDIR)
    if synctool.lib.DRY_RUN:
        msg += ' (dry run)'
    if synctool.lib.OPT_NODENAME:
        msg = ('%s: ' % nodename) + msg
    stdout(msg)

    unix_out(' '.join(dsh_cp_cmd_arr))

    if not synctool.lib.DRY_RUN:
        synctool.lib.run_with_nodename(dsh_cp_cmd_arr, nodename)

def _single_purge_callback(obj, pre_dict, post_dict):
    # type: (SyncObject, Dict[str, str], Dict[str, str]) -> Tuple[bool, bool]
    '''do purge function for single files'''

    # The same as _single_overlay_callback(), except that
    # purge entries may differ in timestamp. synctool has to report
    # this because pure rsync will as well (which is bloody annoying)
    #
    # For normal synctool overlay/, it's regarded as not important
    # and synctool will not complain about it
    #
    # This actually leaves a final wart; synctool --single may create
    # purge entries that rsync will complain about and sync again
    # Anyway, I don't think it's a big deal, and that's what you get
    # when you mix up synctool and rsync

    go_on = True
    updated = False

    if _match_single(obj.dest_path):
        _, updated = _overlay_callback(obj, pre_dict, post_dict)
        if not updated:
            if obj.check_purge_timestamp():
                stdout('%s is up to date' % obj.dest_path)
                terse(synctool.lib.TERSE_OK, obj.dest_path)
                unix_out('# %s is up to date\n' % obj.dest_path)
            # else: pass

        if not SINGLE_FILES:
            return False, updated

    return go_on, updated

    def compare(self, _src_path, dest_stat):
        # type: (str, SyncStat) -> bool
        '''see if devs are the same'''

        if not self.exists:
            return False

        # dest_stat is a SyncStat object and it's useless here
        # I need a real, fresh statbuf that includes st_rdev field
        try:
            dest_stat = os.lstat(self.name)
        except OSError as err:
            error('error checking %s : %s' % (self.name, err.strerror))
            return False

        # Note: mypy triggers false errors here
        # Also, no luck with Union[SyncStat, posix.stat_result]
        # In any case, for VNodeChrDev and VNodeBlkDev,
        # the self.src_stat is of type posix.stat_result
        src_major = os.major(self.src_stat.st_rdev)     # type: ignore
        src_minor = os.minor(self.src_stat.st_rdev)     # type: ignore
        dest_major = os.major(dest_stat.st_rdev)        # type: ignore
        dest_minor = os.minor(dest_stat.st_rdev)        # type: ignore
        if src_major != dest_major or src_minor != dest_minor:
            stdout('%s should have major,minor %d,%d but has %d,%d' %
                   (self.name, src_major, src_minor, dest_major, dest_minor))
            unix_out('# updating major,minor %s' % self.name)
            terse(synctool.lib.TERSE_SYNC, self.name)
            return False

        return True

def _single_overlay_callback(obj, post_dict, updated, *args):
    '''do overlay function for single files'''

    if obj.ov_type == synctool.overlay.OV_TEMPLATE:
        return generate_template(obj, post_dict), False

    go_on = True

    if _match_single(obj.dest_path):
        _, updated = _overlay_callback(obj, post_dict, False, *args)
        if not updated:
            stdout('%s is up to date' % obj.dest_path)
            terse(synctool.lib.TERSE_OK, obj.dest_path)
            unix_out('# %s is up to date\n' % obj.dest_path)
        else:
            # register .post on the parent dir, if it has a .post script
            obj.dest_path = os.path.dirname(obj.dest_path)
            obj.dest_stat = synctool.syncstat.SyncStat(obj.dest_path)

            if obj.dest_path in post_dict:
                changed_dict = args[0]
                changed_dict[obj.dest_path] = (obj, post_dict[obj.dest_path])

        if not SINGLE_FILES:
            return False, updated

    return go_on, updated

def run_command_in_dir(dest_dir, cmd):
    '''change directory to dest_dir, and run the shell command'''

    verbose('  os.chdir(%s)' % dest_dir)
    unix_out('cd %s' % dest_dir)

    cwd = os.getcwd()

    # if dry run, the target directory may not exist yet
    # (mkdir has not been called for real, for a dry run)
    if synctool.lib.DRY_RUN:
        run_command(cmd)

        verbose('  os.chdir(%s)' % cwd)
        unix_out('cd %s' % cwd)
        unix_out('')
        return

    try:
        os.chdir(dest_dir)
    except OSError as err:
        stderr('error changing directory to %s: %s' % (dest_dir,
                                                       err.strerror))
    else:
        run_command(cmd)

        verbose('  os.chdir(%s)' % cwd)
        unix_out('cd %s' % cwd)
        unix_out('')

        try:
            os.chdir(cwd)
        except OSError as err:
            stderr('error changing directory to %s: %s' % (cwd, err.strerror))

    def compare(self, src_path, dest_stat):
        '''see if devs are the same'''

        if not self.exists:
            return False

        # dest_stat is a SyncStat object and it's useless here
        # I need a real, fresh statbuf that includes st_rdev field
        try:
            dest_stat = os.lstat(self.name)
        except OSError as err:
            error('error checking %s : %s' % (self.name, err.strerror))
            return False

        src_major = os.major(self.src_stat.st_rdev)
        src_minor = os.minor(self.src_stat.st_rdev)
        dest_major = os.major(dest_stat.st_rdev)
        dest_minor = os.minor(dest_stat.st_rdev)
        if src_major != dest_major or src_minor != dest_minor:
            stdout('%s should have major,minor %d,%d but has %d,%d' %
                (self.name, src_major, src_minor, dest_major, dest_minor))
            unix_out('# updating major,minor %s' % self.name)
            terse(synctool.lib.TERSE_SYNC, self.name)
            return False

        return True

    def _compare_checksums(self, src_path):
        # type: (str) -> bool
        '''compare checksum of src_path and dest: self.name
        Return True if the same'''

        try:
            f1 = open(src_path, 'rb')
        except IOError as err:
            error('failed to open %s : %s' % (src_path, err.strerror))
            # return True because we can't fix an error in src_path
            return True

        sum1 = hashlib.md5()
        sum2 = hashlib.md5()

        with f1:
            try:
                f2 = open(self.name, 'rb')
            except IOError as err:
                error('failed to open %s : %s' % (self.name, err.strerror))
                return False

            with f2:
                ended = False
                while not ended and (sum1.digest() == sum2.digest()):
                    try:
                        data1 = f1.read(IO_SIZE)
                    except IOError as err:
                        error('failed to read file %s: %s' % (src_path,
                                                              err.strerror))
                        return False

                    if not data1:
                        ended = True
                    else:
                        sum1.update(data1)

                    try:
                        data2 = f2.read(IO_SIZE)
                    except IOError as err:
                        error('failed to read file %s: %s' % (self.name,
                                                              err.strerror))
                        return False

                    if not data2:
                        ended = True
                    else:
                        sum2.update(data2)

        if sum1.digest() != sum2.digest():
            if synctool.lib.DRY_RUN:
                stdout('%s mismatch (MD5 checksum)' % self.name)
            else:
                stdout('%s updated (MD5 mismatch)' % self.name)

            unix_out('# updating file %s' % self.name)
            terse(synctool.lib.TERSE_SYNC, self.name)
            return False

        return True

def ping_node(addr):
    # type: (str) -> None
    '''ping a single node'''

    node = NODESET.get_nodename_from_address(addr)
    verbose('pinging %s' % node)
    unix_out('%s %s' % (param.PING_CMD, addr))

    packets_received = 0

    # execute ping command and show output with the nodename
    cmd = '%s %s' % (param.PING_CMD, addr)
    cmd_arr = shlex.split(cmd)

    try:
        f = subprocess.Popen(cmd_arr, shell=False, bufsize=4096,
                             stdout=subprocess.PIPE,
                             stderr=subprocess.STDOUT).stdout
    except OSError as err:
        error('failed to run command %s: %s' % (cmd_arr[0], err.strerror))
        return

    with f:
        for line in f:
            line = line.strip()

            # argh, we have to parse output here
            #
            # on BSD, ping says something like:
            # "2 packets transmitted, 0 packets received, 100.0% packet loss"
            #
            # on Linux, ping says something like:
            # "2 packets transmitted, 0 received, 100.0% packet loss, " \
            # "time 1001ms"

            arr = line.split()
            if len(arr) > 3 and (arr[1] == 'packets' and
                                 arr[2] == 'transmitted,'):
                try:
                    packets_received = int(arr[3])
                except ValueError:
                    pass

                break

            # some ping implementations say "hostname is alive"
            # or "hostname is unreachable"
            elif len(arr) == 3 and arr[1] == 'is':
                if arr[2] == 'alive':
                    packets_received = 100

                elif arr[2] == 'unreachable':
                    packets_received = -1

    if packets_received > 0:
        print '%s: up' % node
    else:
        print '%s: not responding' % node

def run_local_synctool():
    '''run synctool on the master node itself'''

    cmd_arr = shlex.split(synctool.param.SYNCTOOL_CMD) + PASS_ARGS

    verbose('running synctool on node %s' % synctool.param.NODENAME)
    unix_out(' '.join(cmd_arr))

    synctool.lib.run_with_nodename(cmd_arr, synctool.param.NODENAME)

def _remote_stat(up):
    '''Get stat info of the remote object
    Returns array of RemoteStat data, or None on error
    '''

    # use ssh connection multiplexing (if possible)
    cmd_arr = shlex.split(synctool.param.SSH_CMD)
    use_multiplex = synctool.multiplex.use_mux(up.node)
    if use_multiplex:
        synctool.multiplex.ssh_args(cmd_arr, up.node)

    list_cmd = os.path.join(synctool.param.ROOTDIR, 'sbin',
                            'synctool_list.py')
    cmd_arr.extend(['--', up.address, list_cmd, up.filename])

    verbose('running synctool_list %s:%s' % (up.node, up.filename))
    unix_out(' '.join(cmd_arr))
    try:
        proc = subprocess.Popen(cmd_arr, shell=False, bufsize=4096,
                                stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE)
    except OSError as err:
        error('failed to run command %s: %s' % (cmd_arr[0], err.strerror))
        return None

    out, err = proc.communicate()

    if proc.returncode == 255:
        error('ssh connection to %s failed' % up.node)
        return None
    elif proc.returncode == 127:
        error('remote list command failed')
        return None

    # parse synctool_list output into array of RemoteStat info
    data = []
    for line in out.split('\n'):
        if not line:
            continue

        arr = line.split()
        if arr[0] == 'error:':
            # relay error message
            error(' '.join(arr[1:]))
            return None

        try:
            remote_stat = RemoteStat(arr)
        except ValueError:
            error('unexpected output from synctool_list %s:%s' %
                  (up.node, up.filename))
            return None

        verbose('remote: %r' % remote_stat)
        data.append(remote_stat)

    return data

def worker_synctool(addr):
    '''run rsync of ROOTDIR to the nodes and ssh+synctool, in parallel'''

    nodename = NODESET.get_nodename_from_address(addr)

    if nodename == synctool.param.NODENAME:
        run_local_synctool()
        return

    # rsync ROOTDIR/dirs/ to the node
    # if "it wants it"
    if not (OPT_SKIP_RSYNC or nodename in synctool.param.NO_RSYNC):
        verbose('running rsync $SYNCTOOL/ to node %s' % nodename)
        unix_out('%s %s %s:%s/' % (synctool.param.RSYNC_CMD,
                                   synctool.param.ROOTDIR, addr,
                                   synctool.param.ROOTDIR))

        # make rsync filter to include the correct dirs
        tmp_filename = rsync_include_filter(nodename)

        cmd_arr = shlex.split(synctool.param.RSYNC_CMD)
        cmd_arr.append('--filter=. %s' % tmp_filename)
        cmd_arr.append('--')
        cmd_arr.append('%s/' % synctool.param.ROOTDIR)
        cmd_arr.append('%s:%s/' % (addr, synctool.param.ROOTDIR))

        # double check the rsync destination
        # our filters are like playing with fire
        if not synctool.param.ROOTDIR or (
            synctool.param.ROOTDIR == os.sep):
            stderr('cowardly refusing to rsync with rootdir == %s' %
                   synctool.param.ROOTDIR)
            sys.exit(-1)

        synctool.lib.run_with_nodename(cmd_arr, nodename)

        # delete temp file
        try:
            os.unlink(tmp_filename)
        except OSError:
            # silently ignore unlink error
            pass

    # run 'ssh node synctool_cmd'
    cmd_arr = shlex.split(synctool.param.SSH_CMD)
    cmd_arr.append('--')
    cmd_arr.append(addr)
    cmd_arr.extend(shlex.split(synctool.param.SYNCTOOL_CMD))
    cmd_arr.append('--nodename=%s' % nodename)
    cmd_arr.extend(PASS_ARGS)

    verbose('running synctool on node %s' % nodename)
    unix_out(' '.join(cmd_arr))

    synctool.lib.run_with_nodename(cmd_arr, nodename)

def _exec_diff(src, dest):
    '''execute diff_cmd to display diff between dest and src'''

    verbose('%s %s %s' % (synctool.param.DIFF_CMD, dest, prettypath(src)))
    unix_out('%s %s %s' % (synctool.param.DIFF_CMD, dest, src))

    cmd_arr = shlex.split(synctool.param.DIFF_CMD)
    cmd_arr.append(dest)
    cmd_arr.append(src)

    synctool.lib.exec_command(cmd_arr)

def _remote_isdir(up):
    '''See if the remote rsync source is a directory or a file
    Parameter 'up' is an instance of UploadFile
    Returns: tuple of booleans: (exists, isdir)'''

    cmd_arr = shlex.split(synctool.param.RSYNC_CMD)[:1]
    cmd_arr.append('--list-only')
    cmd_arr.append(up.address + ':' + up.filename)

    verbose('running rsync --list-only %s:%s' % (up.node, up.filename))
    unix_out(' '.join(cmd_arr))

    try:
        proc = subprocess.Popen(cmd_arr, shell=False, bufsize=4096,
                                stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE)
    except OSError as err:
        stderr('failed to run command %s: %s' % (cmd_arr[0], err.strerror))
        return False, False

    out, err = proc.communicate()

    if proc.returncode != 0:
        if proc.returncode == 255:
            stderr('failed to connect to %s' % up.node)
        elif proc.returncode == 23:
            stderr('error: no such file or directory')
        else:
            stderr('failed rsync %s:%s' % (up.node, up.filename))

        return False, False

    # output should be an 'ls -l' like line, with first a mode string
    for line in out.split('\n'):
        arr = line.split()
        mode = arr[0]
        if len(mode) == 10:     # crude test
            if mode[0] == 'd':
                # it's a directory
                verbose('remote rsync source is a directory')
                return True, True

            if mode[0] in '-lpcbs':
                # accept it as a file entry
                verbose('remote rsync source is a file entry')
                return True, False

        # some other line on stdout; just ignore it

    # got no good output
    stderr('failed to parse rsync --list-only output')
    return False, False

def _run_rsync_purge(cmd_arr):
    # type: (List[str]) -> None
    '''run rsync for purging
    cmd_arr holds already prepared rsync command + arguments
    '''

    unix_out(' '.join(cmd_arr))

    sys.stdout.flush()
    sys.stderr.flush()
    try:
        # run rsync
        proc = subprocess.Popen(cmd_arr, shell=False, bufsize=4096,
                                stdout=subprocess.PIPE)
    except OSError as err:
        error('failed to run command %s: %s' % (cmd_arr[0], err.strerror))
        return

    out, _ = proc.communicate()

    if synctool.lib.VERBOSE:
        print out

    out = out.split('\n')
    for line in out:
        line = line.strip()
        if not line:
            continue

        code, filename = line.split(' ', 1)

        if code[:6] == 'ERROR:' or code[:8] == 'WARNING:':
            # output rsync errors and warnings
            stderr(line)
            continue

        if filename == './':
            # rsync has a habit of displaying ugly "./" path
            # cmd_arr[-1] is the destination path
            path = cmd_arr[-1]
        else:
            # cmd_arr[-1] is the destination path
            path = os.path.join(cmd_arr[-1], filename)

        if code[0] == '*':
            # rsync has a message for us
            # most likely "deleting"
            msg = code[1:]
            msg = msg.strip()
            stdout('%s %s (purge)' % (msg, prettypath(path)))
        else:
            stdout('%s mismatch (purge)' % prettypath(path))

    def set_permissions(self):
        '''set access permission bits equal to source'''

        verbose(dryrun_msg('  os.chmod(%s, %04o)' %
                           (self.name, self.stat.mode & 07777)))
        unix_out('chmod 0%o %s' % (self.stat.mode & 07777, self.name))
        if not synctool.lib.DRY_RUN:
            try:
                os.chmod(self.name, self.stat.mode & 07777)
            except OSError as err:
                error('failed to chmod %04o %s : %s' %
                      (self.stat.mode & 07777, self.name, err.strerror))
                terse(synctool.lib.TERSE_FAIL, 'mode %s' % self.name)

    def create(self):
        '''make a fifo'''

        verbose(dryrun_msg('  os.mkfifo(%s)' % self.name))
        unix_out('mkfifo %s' % self.name)
        terse(synctool.lib.TERSE_NEW, self.name)
        if not synctool.lib.DRY_RUN:
            try:
                os.mkfifo(self.name, self.stat.mode & 0777)
            except OSError as err:
                error('failed to create fifo %s : %s' % (self.name,
                                                         err.strerror))
                terse(TERSE_FAIL, 'fifo %s' % self.name)

    def compare(self, src_path, dest_stat):
        '''see if files are the same
        Return True if the same'''

        if self.stat.size != dest_stat.size:
            if synctool.lib.DRY_RUN:
                stdout('%s mismatch (file size)' % self.name)
            else:
                stdout('%s updated (file size mismatch)' % self.name)
            terse(synctool.lib.TERSE_SYNC, self.name)
            unix_out('# updating file %s' % self.name)
            return False

        return self._compare_checksums(src_path)

    def mkdir_basepath(self):
        '''call mkdir -p to create leading path'''

        if synctool.lib.DRY_RUN:
            return

        basedir = os.path.dirname(self.name)

        # be a bit quiet about it
        if synctool.lib.VERBOSE or synctool.lib.UNIX_CMD:
            verbose('making directory %s' % prettypath(basedir))
            unix_out('mkdir -p %s' % basedir)

        synctool.lib.mkdir_p(basedir)

def detect_ssh():
    # type: () -> int
    '''detect ssh version
    Set global SSH_VERSION to 2-digit int number:
    eg. version "5.6p1" -> SSH_VERSION = 56

    Returns: SSH_VERSION
    This routine only works for OpenSSH; otherwise return -1
    '''

    global SSH_VERSION

    if SSH_VERSION is not None:
        return SSH_VERSION

    cmd_arr = shlex.split(synctool.param.SSH_CMD)
    # only use first item: the path to the ssh command
    cmd_arr = cmd_arr[:1]
    cmd_arr.append('-V')

    unix_out(' '.join(cmd_arr))
    try:
        # OpenSSH may print version information on stderr
        proc = subprocess.Popen(cmd_arr, shell=False, stdout=subprocess.PIPE,
                                stderr=subprocess.STDOUT)
    except OSError as err:
        error('failed to execute %s: %s' % (cmd_arr[0], err.strerror))
        SSH_VERSION = -1
        return SSH_VERSION

    # stderr was redirected to stdout
    data, _ = proc.communicate()
    if not data:
        SSH_VERSION = -1
        return SSH_VERSION

    data = data.strip()
    verbose('ssh version string: ' + data)

    # data should be a single line matching "OpenSSH_... SSL ... date\n"
    m = MATCH_SSH_VERSION.match(data)
    if not m:
        SSH_VERSION = -1
        return SSH_VERSION

    groups = m.groups()
    SSH_VERSION = int(groups[0]) * 10 + int(groups[1])
    verbose('SSH_VERSION: %d' % SSH_VERSION)
    return SSH_VERSION

def purge_files():
    '''run the purge function'''

    paths = []
    purge_groups = os.listdir(synctool.param.PURGE_DIR)

    # find the source purge paths that we need to copy
    # scan only the group dirs that apply
    for g in synctool.param.MY_GROUPS:
        if g in purge_groups:
            purge_root = os.path.join(synctool.param.PURGE_DIR, g)
            if not os.path.isdir(purge_root):
                continue

            for path, subdirs, files in os.walk(purge_root):
                # rsync only purge dirs that actually contain files
                # otherwise rsync --delete would wreak havoc
                if not files:
                    continue

                if path == purge_root:
                    # root contains files; guard against user mistakes
                    # rsync --delete would destroy the whole filesystem
                    stderr('cowardly refusing to purge the root directory')
                    stderr('please remove any files directly under %s/' %
                            prettypath(purge_root))
                    return

                # paths has (src_dir, dest_dir)
                paths.append((path, path[len(purge_root):]))

                # do not recurse into this dir any deeper
                del subdirs[:]

    cmd_rsync, opts_string = _make_rsync_purge_cmd()

    # call rsync to copy the purge dirs
    for src, dest in paths:
        # trailing slash on source path is important for rsync
        src += os.sep
        dest += os.sep

        cmd_arr = cmd_rsync[:]
        cmd_arr.append(src)
        cmd_arr.append(dest)

        verbose('running rsync%s%s %s' % (opts_string, prettypath(src), dest))
        unix_out(' '.join(cmd_arr))
        _run_rsync_purge(cmd_arr)