path: root/src/inode2filename/search_directories.cc

// Copyright (C) 2018 The Android Open Source Project
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "common/debug.h"
#include "inode2filename/search_directories.h"
#include "inode2filename/system_call.h"

#include <android-base/file.h>
#include <android-base/logging.h>
#include <android-base/scopeguard.h>
#include <android-base/stringprintf.h>
#include <android-base/unique_fd.h>

#include "rxcpp/rx.hpp"

#include <iostream>
#include <stdio.h>
#include <fstream>
#include <vector>
#include <optional>

#include <signal.h>
#include <stdlib.h>
#include <unistd.h>

#include <sys/types.h>

#ifdef __ANDROID__
#include <sys/sysmacros.h>
#endif

#include <sys/stat.h>
#include <fcntl.h>
#include <poll.h>
#include <dirent.h>

#include <unordered_map>

namespace rx = rxcpp;
using android::base::unique_fd;  // NOLINT
using android::base::StringPrintf;  // NOLINT

namespace iorap::inode2filename {

// A multimap of 'ino_t -> List[Inode]' (where the value Inodes have the same ino_t as the key).
//
// A flat list of Inodes is turned into the above map, then keys can be removed one at a time
// until the InodeSet eventually becomes empty.
struct InodeSet {
  struct ValueRange {
    auto/*Iterable<Inode>*/ begin() {
      return begin_;
    }

    auto/*Iterable<Inode>*/ end() {
      return end_;
    }

    bool empty() const {
      return begin_ == end_;
    }

    explicit operator bool() const {
      return !empty();
    }

    std::unordered_multimap<ino_t, Inode>::iterator begin_, end_;

    friend std::ostream& operator<<(std::ostream& os, const ValueRange& s);
  };

  // Create an observable that emits the remaining inodes in the map.
  //
  // Mutation functions must not be called until this observable
  // has been finished emitting all values (e.g. with on_completed) since that
  // would cause the underlying iterators to go into an undefined state.
  auto/*observable<Inode>*/ IterateValues() const {
    return rxcpp::observable<>::iterate(set_).map(  // XX: should we use identity_immediate here?
        [](const std::pair<const ino_t, Inode>& pair) {
          return pair.second;
        }
    );
    // TODO: this would be more efficient as a range-v3 view.
  }

  constexpr bool Empty() const {
    return set_.empty();
  }

  static InodeSet OfList(const std::vector<Inode>& list) {
    InodeSet new_inode_set;
    std::unordered_multimap<ino_t, Inode>* map = &new_inode_set.set_;

    for (const Inode& inode : list) {
      map->insert({inode.inode, inode});
    }

    return new_inode_set;
  }

  // Return an optional list of 'Inode' structs whose 'inode' field matches the 'inode' parameter.
  // Returns an empty range if there was nothing found.
  ValueRange FindInodeList(ino_t inode) {
    auto range = set_.equal_range(inode);
    return ValueRange{range.first, range.second};
  }

  // Match all fields of an Inode against a 'struct stat' stat_buf.
  //
  // The returned Inode (if any) is removed from the InodeSet; it will not be returned by
  // FindInodeList in future calls.
  std::optional<Inode> FindAndRemoveInodeInList(ValueRange inode_list,
                                                const struct stat& stat_buf) {
    LOG(VERBOSE) << "FindAndRemoveInodeInList " << inode_list << ", "
                 << "stat_buf{st_dev=" << stat_buf.st_dev << ",st_ino=" << stat_buf.st_ino << "}";

    auto /*iterator*/ found = std::find_if(inode_list.begin(),
                                           inode_list.end(),
                                           [&](const std::pair<ino_t, Inode>& pair) {
      const Inode& inode = pair.second;
      if (inode.inode != stat_buf.st_ino) {
        return false;
      }

      dev_t inode_dev =
          makedev(static_cast<int>(inode.device_major), static_cast<int>(inode.device_minor));

      // Inodes could be the same across different devices.
      // Also match the device id.
      if (inode_dev != stat_buf.st_dev) {
        LOG(VERBOSE) << "InodeSet:FindAndRemoveInodeInList matched ino: " << inode.inode
                     << " but not device"
                     << ", expected dev: " << stat_buf.st_dev
                     << ", actual dev: " << inode_dev;
        return false;
      }
      return true;
    });

    if (found != inode_list.end()) {
      const Inode& inode = found->second;
      LOG(VERBOSE) << "InodeSet:FindAndRemoveInodeInList *success* inode+device " << inode;
      DCHECK(found->second.inode == stat_buf.st_ino);
      // Erase the inode from the list. This is important.
      set_.erase(found);
      return inode;
    }

    return std::nullopt;
  }

  // TODO: equality and string operators for testing/logging.
 private:
  // Explanation: readdir returns a 'file' -> 'ino_t inode' mapping.
  //
  // However inodes can be reused on different partitions (but they have a different device number).
  // To handle this edge case, and to avoid calling stat whenever the inode definitely doesn't match
  // store the inodes into a single-key,multi-value container.
  //
  // This enables fast scanning of readdir results by matching just the 'inode' portion,
  // then calling stat only when the inode portion definitely matches to confirm the device.

  // There are no single-key multi-value containers in standard C++, so pretend
  // we have one by writing this simple facade around an unordered set.
  //
  // We expect that the vector size is usually size=1 (or 2 or 3) since the # of devices
  // is fixed by however many partitions there are on the system, AND the same inode #
  // would have to be reused across a different file.
  std::unordered_multimap<ino_t, Inode> set_;  // TODO: Rename to map_.

  friend std::ostream& operator<<(std::ostream& os, const InodeSet& s);
};

std::ostream& operator<<(std::ostream& os, const InodeSet& s) {
  os << "InodeSet{";
  for (const auto& kv : s.set_) {
    // e.g. "123=>(1:2:123)" ... its expected for the 'ino_t' portion to be repeated.
    os << "" << kv.first << "=>(" << kv.second << "),";
  }
  os << "}";
  return os;
}

std::ostream& operator<<(std::ostream& os, const InodeSet::ValueRange& v) {
  // Don't want to make a const and non const version of ValueRange.
  InodeSet::ValueRange& s = const_cast<InodeSet::ValueRange&>(v);

  os << "InodeSet::ValueRange{";
  for (const auto& kv : s) {
    // e.g. "123=>(1:2:123)" ... its expected for the 'ino_t' portion to be repeated.
    os << "" << kv.first << "=>(" << kv.second << "),";
  }
  os << "}";
  return os;
}

void search_for_inodes_in(std::vector<Inode>& inode_list, const std::string& dirpath);

enum DirectoryEntryErrorCode {
  kInvalid,    // not a real error code. to detect bad initialization.
  kOpenDir,    // opendir failed.
  kReadDir,    // readdir failed.
  kDtUnknown,  // d_type was DT_UNKNOWN error.
};

struct DirectoryEntryError {
  DirectoryEntryErrorCode code;
  int err_no;
  std::string filename;
};

std::ostream& operator<<(std::ostream& os, const DirectoryEntryError& e) {
  os << "DirectoryEntryError{"
     << static_cast<int>(e.code) << "," << e.err_no << "," << e.filename << "}";
  return os;
  // TODO: pretty-print code and err-no
}

static common::DebugCounter gDebugDirectoryEntryCounter{};
static constexpr bool kDebugDirectoryEntry = false;

#define DIRECTORY_ENTRY_MOVE_DCHECK() \
    DCHECK_EQ(other.moved_from_, false) << __PRETTY_FUNCTION__ << "CNT:" << other.debug_counter_;
#define DIRECTORY_ENTRY_TRACE_CTOR() \
    if (kDebugDirectoryEntry) LOG(VERBOSE) << __PRETTY_FUNCTION__ << "@CNT:" << debug_counter_

struct DirectoryEntry {
  using ResultT = iorap::expected<DirectoryEntry, DirectoryEntryError>;
  using ObservableT = rx::observable<ResultT>;

  static constexpr ino_t kInvalidIno = std::numeric_limits<ino_t>::max();
  static constexpr auto kInvalidFileName = "";

  // Path to file, the prefix is one of the root directories.
  std::string filename{kInvalidFileName};
  // Inode number of the file. Not unique across different devices.
  ino_t d_ino{kInvalidIno};
  // File type (DT_LNK, DT_REG, DT_DIR, or DT_UNKNOWN)
  unsigned char d_type{DT_UNKNOWN};  // Note: not seen outside of sentinel roots.
  // TODO: Consider invariant checks for valid combinations of above fields?

  // Debug-only flags.
  bool moved_from_{false};
  size_t debug_counter_{0};

 private:
  // TODO: remove default constructor?
  //
  // SEEMS TO BE USED by std::vector etc. FIX DAT.
  DirectoryEntry() noexcept {
    debug_counter_ = gDebugDirectoryEntryCounter++;
    DIRECTORY_ENTRY_TRACE_CTOR();
  }
 public:
  DirectoryEntry(std::string filename, ino_t d_ino, unsigned char d_type) noexcept
    : filename{std::move(filename)},
      d_ino{d_ino},
      d_type{d_type} {
    debug_counter_ = gDebugDirectoryEntryCounter++;
    DIRECTORY_ENTRY_TRACE_CTOR();
  }

  DirectoryEntry(const DirectoryEntry& other) noexcept {
    // Do not use member-initialization syntax so that this DCHECK can execute first.
    DIRECTORY_ENTRY_MOVE_DCHECK();

    filename = other.filename;
    d_ino = other.d_ino;
    d_type = other.d_type;
    children_paths_ = other.children_paths_;
    children_initialized_ = other.children_initialized_;
    debug_counter_ = other.debug_counter_;
    DIRECTORY_ENTRY_TRACE_CTOR();
  }

  DirectoryEntry& operator=(const DirectoryEntry& other) noexcept {
    if (this == &other) {
      return *this;
    }

    DIRECTORY_ENTRY_MOVE_DCHECK();

    filename = other.filename;
    d_ino = other.d_ino;
    d_type = other.d_type;
    children_paths_ = other.children_paths_;
    children_initialized_ = other.children_initialized_;
    debug_counter_ = other.debug_counter_;
    DIRECTORY_ENTRY_TRACE_CTOR();

    return *this;
  }

  DirectoryEntry& operator=(DirectoryEntry&& other) noexcept {
    if (this == &other) {
      return *this;
    }

    DIRECTORY_ENTRY_MOVE_DCHECK();

    filename = std::move(other.filename);
    d_ino = other.d_ino;
    d_type = other.d_type;
    children_paths_ = std::move(other.children_paths_);
    children_initialized_ = other.children_initialized_;
    debug_counter_ = other.debug_counter_;
    DIRECTORY_ENTRY_TRACE_CTOR();

    return *this;
  }

  DirectoryEntry(DirectoryEntry&& other) noexcept {
    DIRECTORY_ENTRY_MOVE_DCHECK();
    other.moved_from_ = true;

    filename = std::move(other.filename);
    d_ino = other.d_ino;
    d_type = other.d_type;
    children_paths_ = std::move(other.children_paths_);
    children_initialized_ = other.children_initialized_;
    debug_counter_ = other.debug_counter_;
    DIRECTORY_ENTRY_TRACE_CTOR();
  }

  // Create a sentinel (root of roots) whose children entries are those specified by
  // children_paths.
  static DirectoryEntry CreateSentinel(std::vector<std::string> children_paths) {
    DirectoryEntry e;
    e.d_type = DT_DIR;
    ++gDebugDirectoryEntryCounter;

    for (std::string& child_path : children_paths) {
      // TODO: Should we call Stat on the child path here to reconstitute the ino_t for a root dir?
      // Otherwise it can look a little strange (i.e. the root dir itself will never match
      // the searched inode).
      //
      // Probably not too big of a problem in practice.
      DirectoryEntry child_entry{std::move(child_path), kInvalidIno, DT_DIR};
      ResultT child_entry_as_result{std::move(child_entry)};
      e.children_paths_.push_back(std::move(child_entry_as_result));
    }

    e.children_initialized_ = true;

    return e;
  }

  // Return an observable which emits the direct children only.
  // The children entries are now read from disk (with readdir) if they weren't read previously.
  std::vector<ResultT> GetChildrenEntries(borrowed<SystemCall*> system_call) const& {
    BuildChildrenPaths(system_call);
    return children_paths_;
  }

  // Return an observable which emits the direct children only.
  // The children entries are now read from disk (with readdir) if they weren't read previously.
  // Movable overload.
  std::vector<ResultT> GetChildrenEntries(borrowed<SystemCall*> system_call) && {
    BuildChildrenPaths(system_call);
    return std::move(children_paths_);
  }

  // Returns a (lazy) observable that emits every single node, in pre-order,
  // rooted at this tree.
  //
  // New entries are only read from disk (with e.g. readdir) when more values are pulled
  // from the observable. Only the direct children of any entry are read at any time.
  //
  // The emission can be stopped prematurely by unsubscribing from the observable.
  // This means the maximum amount of 'redundant' IO reads is bounded by the children count
  // of all entries emitted thus far minus entries actually emitted.
  ObservableT GetSubTreePreOrderEntries(borrowed<SystemCall*> system_call) const;

 private:
  // Out-of-line definition to avoid circular type dependency.
  void BuildChildrenPaths(borrowed<SystemCall*> system_call) const;

  // We need to lazily initialize children_paths_ only when we try to read them.
  //
  // Assuming the underlying file system doesn't change (which isn't strictly true),
  // the directory children are referentially transparent.
  //
  // In practice we do not need to distinguish between the file contents changing out
  // from under us in this code, so we don't need the more strict requirements.
  mutable std::vector<ResultT> children_paths_;
  mutable bool children_initialized_{false};

  friend std::ostream& operator<<(std::ostream& os, const DirectoryEntry& d);
};

std::ostream& operator<<(std::ostream& os, const DirectoryEntry& d) {
  os << "DirectoryEntry{" << d.filename << ",ino:" << d.d_ino << ",type:" << d.d_type << "}";
  return os;
}

using DirectoryEntryResult = DirectoryEntry::ResultT;

// Read all directory entries and return it as a vector. This must be an eager operation,
// as readdir is not re-entrant.
//
// This could be considered as a limitation from the 'observable' perspective since
// one can end up reading unnecessary extra directory entries that are then never consumed.
//
// The following entries are skipped:
//  - '.' self
//  - ".." parent
//
// All DT types except the following are removed:
//  * DT_LNK - symbolic link (empty children)
//  * DT_REG - regular file  (empty children)
//  * DT_DIR - directory     (has children)
static std::vector<DirectoryEntryResult>
    ReadDirectoryEntriesFromDirectoryPath(std::string dirpath, borrowed<SystemCall*> system_call) {
  DIR *dirp;
  struct dirent *dp;

  LOG(VERBOSE) << "ReadDirectoryEntriesFromDirectoryPath(" << dirpath << ")";

  if ((dirp = system_call->opendir(dirpath.c_str())) == nullptr) {
    PLOG(ERROR) << "Couldn't open directory: " << dirpath;
    return {DirectoryEntryError{kOpenDir, errno, dirpath}};
  }

  // Read all the results up front because readdir is not re-entrant.
  std::vector<DirectoryEntryResult> results;

  // Get full path + the directory entry path.
  auto child_path = [&] { return dirpath + "/" + dp->d_name; };

  do {
    errno = 0;
    if ((dp = system_call->readdir(dirp)) != nullptr) {
      if (dp->d_type == DT_DIR) {
        if (strcmp(".", dp->d_name) == 0 || strcmp("..", dp->d_name) == 0) {
          LOG(VERBOSE) << "Skip self/parent: " << dp->d_name;
          continue;
        }

        LOG(VERBOSE) << "Find entry " << child_path()
                     << ", ino: " << dp->d_ino << ", type: " << dp->d_type;
        results.push_back(DirectoryEntry{child_path(),
                                         static_cast<ino_t>(dp->d_ino),
                                         dp->d_type});
      } else if (dp->d_type == DT_UNKNOWN) {
        // This seems bad if it happens. We should probably do something about this.
        LOG(WARNING) << "Found unknown DT entry: " << child_path();

        results.push_back(DirectoryEntryError{kDtUnknown, /*errno*/0, child_path()});
      } else if (dp->d_type == DT_LNK || dp->d_type == DT_REG) {
        // Regular non-directory file entry.
        results.push_back(DirectoryEntry{child_path(),
                                         static_cast<ino_t>(dp->d_ino),
                                         dp->d_type});
      } else {
        // Block device, character device, socket, etc...
        LOG(VERBOSE) << "Skip DT entry of type: " << dp->d_type << " " << child_path();
      }
    } else if (errno != 0) {
      PLOG(ERROR) << "Error reading directory entry in " << dirpath;

      results.push_back(DirectoryEntryError{kReadDir, errno, dirpath});
    }
  } while (dp != nullptr);

  if (system_call->closedir(dirp) < 0) {
    PLOG(ERROR) << "Failed to close directory " << dirpath;
  }

  return results;
}

void DirectoryEntry::BuildChildrenPaths(borrowed<SystemCall*> system_call) const {
  if (children_initialized_) {
    return;
  }

  if (d_type == DT_DIR) {
    children_paths_ = ReadDirectoryEntriesFromDirectoryPath(filename, system_call);
    // TODO: consider using dependency injection here to substitute this function during testing?
  }
}

struct InodeSearchParameters {
  std::vector<Inode> inode_list;
  std::vector<std::string> root_dirs;
};

// [IN]
// observable: expected<Value, Error>, ...
// [OUT]
// observable: Value, ...
//
// Any encountered 'Error' items are dropped after logging.
template <typename T>
auto MapExpectedOrLog(T&& observable,
                      ::android::base::LogSeverity log_level) {
  return observable.filter([log_level](const auto& result) {
    if (result) {
      return true;
    } else {
      LOG(log_level) << result.error();
      return false;
    }
  }).map([](auto&& result) {
    return IORAP_FORWARD_LAMBDA(result).value();
  });
}

template <typename T>
auto MapExpectedOrLogError(T&& observable) {
  return MapExpectedOrLog(std::forward<T>(observable), ::android::base::ERROR);
}

template <typename T>
auto MapOptionalOrDrop(T&& observable) {
  return observable.filter([](const auto& result) {
    return result.has_value();
  }).map([](auto&& result) {
    return IORAP_FORWARD_LAMBDA(result).value();
  });
  // TODO: static_assert this isn't used with an unexpected.
}

template <typename T, typename F>
auto VisitValueOrLogError(T&& expected, F&& visit_func, const char* error_prefix = "") {
  if (!expected) {
    LOG(ERROR) << error_prefix << " " << expected.error();
  } else {
    visit_func(std::forward<T>(expected).value());
  }
  // TODO: Could be good to make this more monadic by returning an optional.
}

template <typename TSimple, typename T, typename F>
void TreeTraversalPreOrderObservableImpl(rx::subscriber<TSimple> dest, T&& node, F&& fn) {
  LOG(VERBOSE) << "TreeTraversalPreOrderObservableImpl (begin) " << __PRETTY_FUNCTION__;

  if (!dest.is_subscribed()) {
    LOG(VERBOSE) << "TreeTraversalPreOrderObservableImpl (unsubscribed)";
    return;
  } else {
    LOG(VERBOSE) << "TreeTraversalPreOrderObservableImpl (on_next node)";

    // Copy the node here. This is less bad than it seems since we haven't yet
    // calculated its children (except in the root), so its just doing a shallow memcpy (sizeof(T)).
    //
    // This assumes the children are calculated lazily, otherwise we'd need to have a separate
    // NodeBody class which only holds the non-children elements.

    TSimple copy = std::forward<T>(node);
    dest.on_next(std::move(copy));

    if (!node.has_value()) {
      return;
    }

    // Whenever we call 'on_next' also check if we end up unsubscribing.
    // This avoids the expensive call into the children.
    if (!dest.is_subscribed()) {
      LOG(VERBOSE) << "TreeTraversalPreOrderObservableImpl (post-self unsubscribe)";
      return;
    }

    // Eagerly get the childrem, moving them instead of copying them.
    auto&& children = fn(std::forward<T>(node));
    for (auto&& child : children) {
      TreeTraversalPreOrderObservableImpl(dest, IORAP_FORWARD_LAMBDA(child), fn);
      // TODO: double check this is doing the std::move properly for rvalues.

      if (!dest.is_subscribed()) {
        LOG(VERBOSE) << "TreeTraversalPreOrderObservableImpl (unsubscribed in children)";
        break;
      }
    };
  }
}

// Creates an observable over all the nodes in the tree rooted at node.
// fn is a function that returns the children of that node.
//
// The items are emitted left-to-right pre-order, and stop early if the
// observable is unsubscribed from.
//
// Implementation requirement:
//    typeof(node) -> expected<V, E> or optional<V> or similar.
//    fn(node) -> iterable<typeof(node)>
//
// preorder(self):
//   visit(self)
//   for child in fn(self):
//     preorder(child)
template <typename T, typename F>
auto/*observable<T>*/ TreeTraversalPreOrderObservable(T&& node, F&& fn) {
  LOG(VERBOSE) << "TreeTraversalPreOrderObservable: " << __PRETTY_FUNCTION__;

  using T_simple = std::decay_t<T>;
  return rx::observable<>::create<T_simple>(
    // Copy node to avoid lifetime issues.
    [node=node,fn=std::forward<F>(fn)](rx::subscriber<T_simple> dest) {
      LOG(VERBOSE) << "TreeTraversalPreOrderObservable (lambda)";
      TreeTraversalPreOrderObservableImpl<T_simple>(dest,
                                                    std::move(node),
                                                    std::move(fn));
      dest.on_completed();
    }
  );
}

DirectoryEntry::ObservableT
    DirectoryEntry::GetSubTreePreOrderEntries(borrowed<SystemCall*> system_call) const {
  return TreeTraversalPreOrderObservable(
      DirectoryEntryResult{*this},
      [system_call=system_call](auto/*DirectoryEntryResult*/&& result)
          -> std::vector<DirectoryEntryResult> {
        if (!result) {
          LOG(VERBOSE) << "GetSubTreePreOrderEntries (no value return)";
          // Cannot have children when it was an error.
          return {};
        }
        return
            IORAP_FORWARD_LAMBDA(result)
            .value()
            .GetChildrenEntries(system_call);
      });
}

struct StatError {
  int err_no;
  std::string path_name;
};

std::ostream& operator<<(std::ostream& os, const StatError& e) {
  os << "StatError{" << e.err_no << "," << e.path_name << "}";
  return os;
}

template <typename U = void>  // suppress unused warning.
static iorap::expected<struct stat, StatError> Stat(const std::string& path_name,
                                                    borrowed<SystemCall*> system_call) {
  struct stat statbuf{};

  // Call stat(2) in live code. Overridden in test code.
  if (system_call->stat(path_name.c_str(), /*out*/&statbuf) == 0) {
    return statbuf;
  } else {
    return iorap::unexpected(StatError{errno, path_name});
  }
}

using StatResult = iorap::expected<struct stat, StatError>;

// An inode's corresponding filename on the system.
struct SearchMatch {
  Inode inode;
  // Relative path joined with a root directory.
  //
  // Use absolute path root dirs to get back absolute path filenames.
  // If relative, this is relative to the current working directory.
  std::string filename;
};

std::ostream& operator<<(std::ostream& os, const SearchMatch& s) {
  os << "SearchMatch{" << s.inode << ", " << s.filename << "}";
  return os;
}

struct SearchState {
  // Emit 'match' Inodes corresponding to the ones here.
  InodeSet inode_set;

  // An inode matching one of the ones in inode_set was discovered in the most-recently
  // emitted SearchState.
  //
  // The InodeSet removes any matching 'Inode'.
  std::optional<SearchMatch> match;

  // TODO: make sure this doesn't copy [inodes], as that would be unnecessarily expensive.
};

std::ostream& operator<<(std::ostream& os, const SearchState& s) {
  os << "SearchState{match:";
  // Print the 'match' first. The InodeSet could be very large so it could be truncated in logs.
  if (s.match) {
    os << s.match.value();
  } else {
    os << "(none)";
  }
  os << ", inode_set:" << s.inode_set << "}";
  return os;
}

// TODO: write operator<< etc.

// Return a lazy observable that will search for all filenames whose inodes
// match the inodes in inode_search_list.
//
// Every unmatched inode will be emitted as an unexpected at the end of the stream.
auto/*[observable<InodeResult>, connectable]*/ SearchDirectoriesForMatchingInodes(
    std::vector<std::string> root_dirs,
    std::vector<Inode> inode_search_list,
    borrowed<SystemCall*> system_call) {

  // Create a (lazy) observable that will emit each DirectoryEntry that is a recursive subchild
  // of root_dirs. Emission will be stopped when its unsubscribed from.
  //
  // This is done by calling readdir(3) lazily.
  auto/*obs<DirectoryEntry>*/ find_all_subdir_entries = ([&]() {
    DirectoryEntry sentinel = DirectoryEntry::CreateSentinel(std::move(root_dirs));
    auto/*obs<DirectoryEntryResult*/ results = sentinel.GetSubTreePreOrderEntries(system_call);

    // Drop any errors by logging them to logcat. "Unwrap" the expected into the underlying data.
    auto/*obs<DirectoryEntry*>*/ expected_drop_errors = MapExpectedOrLogError(std::move(results));
    return expected_drop_errors;
  })();

  // DirectoryEntry is missing the dev_t portion, so we may need to call scan(2) again
  // to confirm the dev_t. We skip calling scan(2) when the ino_t does not match.
  // InodeSet lets us optimally avoid calling scan(2).
  SearchState initial;
  initial.inode_set = InodeSet::OfList(inode_search_list);

  auto/*[observable<SearchState>,Connectable]*/ search_state_results = find_all_subdir_entries.scan(
      std::move(initial),
      [system_call=system_call](SearchState search_state, const DirectoryEntry& dir_entry) {
        LOG(VERBOSE) << "SearchDirectoriesForMatchingInodes#Scan "
                     << dir_entry << ", state: " << search_state;

        search_state.match = std::nullopt;

        InodeSet* inodes = &search_state.inode_set;

        // Find all the possible inodes across different devices.
        InodeSet::ValueRange inode_list = inodes->FindInodeList(dir_entry.d_ino);

        // This directory doesn't correspond to any inodes we are searching for.
        if (!inode_list) {
          return search_state;
        }

        StatResult maybe_stat = Stat(dir_entry.filename, system_call);
        VisitValueOrLogError(maybe_stat, [&](const struct stat& stat_buf) {
          // Try to match the specific inode. Usually this will not result in a match (nullopt).
          std::optional<Inode> inode = inodes->FindAndRemoveInodeInList(inode_list, stat_buf);

          if (inode) {
            search_state.match = SearchMatch{inode.value(), dir_entry.filename};
          }
        });

        return search_state;  // implicit move.
      }
  // Avoid exhausting a potentially 'infinite' stream of files by terminating as soon
  // as we find every single inode we care about.
  ).take_while([](const SearchState& state) {
      // Also emit the last item that caused the search set to go empty.
      bool cond = !state.inode_set.Empty() || state.match;

      if (WOULD_LOG(VERBOSE)) {
        static int kCounter = 0;
        LOG(VERBOSE) << "SearchDirectoriesForMatchingInodes#take_while (" << kCounter++ <<
                     ",is_empty:"
                     << state.inode_set.Empty() << ", match:" << state.match.has_value();
      }
      // Minor O(1) implementation inefficiency:
      // (Too minor to fix but it can be strange if looking at the logs or readdir traces).
      //
      // Note, because we return 'true' after the search set went empty,
      // the overall stream graph still pulls from search_state_results exactly once more:
      //
      // This means that for cond to go to false, we would've read one extra item and then discarded
      // it. If that item was the first child of a directory, that means we essentially did
      // one redundant pass of doing a readdir.
      //
      // In other words if the search set goes to empty while the current item is a directory,
      // it will definitely readdir on it at least once as we try to get the first child in
      // OnTreeTraversal.
      //
      // This could be fixed with a 'take_until(Predicate)' operator which doesn't discard
      // the last item when the condition becomes false. However rxcpp seems to lack this operator,
      // whereas RxJava has it.

      if (!cond) {
        LOG(VERBOSE) << "SearchDirectoriesForMatchingInodes#take_while "
                     << "should now terminate for " << state;
      }

      return cond;
  }).publish();
  // The publish here is mandatory. The stream is consumed twice (once by matched and once by
  // unmatched streams). Without the publish, once all items from 'matched' were consumed it would
  // start another instance of 'search_state_results' (i.e. it appears as if the search
  // is restarted).
  //
  // By using 'publish', the search_state_results is effectively shared by both downstream nodes.
  // Note that this also requires the subscriber to additionally call #connect on the above stream,
  // otherwise no work will happen.

  // Lifetime notes:
  //
  // The the 'SearchState' is emitted into both below streams simultaneously.
  //    The 'unmatched_inode_values' only touches the inode_set.
  //    The 'matched_inode_values' only touches the match.
  // Either stream can 'std::move' from those fields because they don't move each other's fields.
  auto/*observable<InodeResult>*/ matched_inode_values = search_state_results
      .filter([](const SearchState& search_state) { return search_state.match.has_value(); })
      .map([](SearchState& search_state) { return std::move(search_state.match.value()); })
                     // observable<SearchMatch>
      .map([](SearchMatch search_match) {
          return InodeResult::makeSuccess(search_match.inode, std::move(search_match.filename));
      });            // observable<InodeResult>

  auto/*observable<?>*/ unmatched_inode_values = search_state_results
      // The 'last' SearchState is the one that contains all the remaining inodes.
      .take_last(1)  // observable<SearchState>
      .flat_map([](const SearchState& search_state) {
          LOG(VERBOSE) << "SearchDirectoriesForMatchingInodes#unmatched -- flat_map";
          // Aside: Could've used a move here if the inodes weren't so lightweight already.
          return search_state.inode_set.IterateValues(); })
                     // observable<Inode>
      .map([](const Inode& inode) {
          LOG(VERBOSE) << "SearchDirectoriesForMatchingInodes#unmatched -- map";
          return InodeResult::makeFailure(inode, InodeResult::kCouldNotFindFilename);
      });
                     // observable<InodeResult>

  // The matched and unmatched InodeResults are emitted together.
  //   Use merge, not concat, because we need both observables to be subscribed to simultaneously.

  auto/*observable<InodeResult*/ all_inode_results =
      matched_inode_values.merge(unmatched_inode_values);

  // Now that all mid-stream observables have been connected, turn the Connectable observable
  // into a regular observable.

  // The caller has to call 'connect' on the search_state_results after subscribing
  // and before any work can actually start.
  return std::make_pair(all_inode_results, search_state_results);
}


rxcpp::observable<InodeResult> SearchDirectories::FindFilenamesFromInodes(
    std::vector<std::string> root_directories,
    std::vector<Inode> inode_list,
    SearchMode mode) {
  DCHECK(mode == SearchMode::kInProcessDirect) << " other modes not implemented yet";

  auto/*observable[2]*/ [inode_results, connectable] = SearchDirectoriesForMatchingInodes(
      std::move(root_directories),
      std::move(inode_list),
      system_call_);

  return inode_results;
}

// I think we could avoid this with auto_connect, which rxcpp doesn't seem to have.
//
// I can't figure out any other way to avoid this, or at least to allow connecting
// on the primary observable (instead of a secondary side-observable).
//
// If using the obvious publish+ref_count then the unmerged stream gets no items emitted into it.
// If tried to ref_count later, everything turns into no-op.
// If trying to call connect too early, the subscribe is missed.
template <typename T>
struct RxAnyConnectableFromObservable : public SearchDirectories::RxAnyConnectable {
  virtual void connect() override {
    observable.connect();
  }

  virtual ~RxAnyConnectableFromObservable() {}

  RxAnyConnectableFromObservable(rxcpp::connectable_observable<T> observable)
    : observable(observable) {
  }

  rxcpp::connectable_observable<T> observable;
};

// Type deduction helper.
template <typename T>
std::unique_ptr<SearchDirectories::RxAnyConnectable>
    MakeRxAnyConnectableFromObservable(rxcpp::connectable_observable<T> observable) {
  SearchDirectories::RxAnyConnectable* ptr = new RxAnyConnectableFromObservable<T>{observable};
  return std::unique_ptr<SearchDirectories::RxAnyConnectable>{ptr};
}

std::pair<rxcpp::observable<InodeResult>, std::unique_ptr<SearchDirectories::RxAnyConnectable>>
    SearchDirectories::FindFilenamesFromInodesPair(
        std::vector<std::string> root_directories,
        std::vector<Inode> inode_list,
        SearchMode mode) {
  DCHECK(mode == SearchMode::kInProcessDirect) << " other modes not implemented yet";

  auto/*observable[2]*/ [inode_results, connectable] = SearchDirectoriesForMatchingInodes(
      std::move(root_directories),
      std::move(inode_list),
      system_call_);

  std::unique_ptr<SearchDirectories::RxAnyConnectable> connectable_ptr =
    MakeRxAnyConnectableFromObservable(connectable.as_dynamic());

  return {inode_results, std::move(connectable_ptr)};
}

}  // namespace iorap::inode2filename