[Summary view] [Print] [Text view]

   1  <?php
   2  
   3  /**
   4   * Add and remove edges between objects. You can use
   5   * @{class:PhabricatorEdgeQuery} to load object edges. For more information
   6   * on edges, see @{article:Using Edges}.
   7   *
   8   * Edges are not directly policy aware, and this editor makes low-level changes
   9   * below the policy layer.
  10   *
  11   *    name=Adding Edges
  12   *    $src  = $earth_phid;
  13   *    $type = PhabricatorEdgeConfig::TYPE_BODY_HAS_SATELLITE;
  14   *    $dst  = $moon_phid;
  15   *
  16   *    id(new PhabricatorEdgeEditor())
  17   *      ->addEdge($src, $type, $dst)
  18   *      ->save();
  19   *
  20   * @task edit     Editing Edges
  21   * @task cycles   Cycle Prevention
  22   * @task internal Internals
  23   */
  24  final class PhabricatorEdgeEditor extends Phobject {
  25  
  26    private $addEdges = array();
  27    private $remEdges = array();
  28    private $openTransactions = array();
  29  
  30  
  31  /* -(  Editing Edges  )------------------------------------------------------ */
  32  
  33  
  34    /**
  35     * Add a new edge (possibly also adding its inverse). Changes take effect when
  36     * you call @{method:save}. If the edge already exists, it will not be
  37     * overwritten, but if data is attached to the edge it will be updated.
  38     * Removals queued with @{method:removeEdge} are executed before
  39     * adds, so the effect of removing and adding the same edge is to overwrite
  40     * any existing edge.
  41     *
  42     * The `$options` parameter accepts these values:
  43     *
  44     *   - `data` Optional, data to write onto the edge.
  45     *   - `inverse_data` Optional, data to write on the inverse edge. If not
  46     *     provided, `data` will be written.
  47     *
  48     * @param phid  Source object PHID.
  49     * @param const Edge type constant.
  50     * @param phid  Destination object PHID.
  51     * @param map   Options map (see documentation).
  52     * @return this
  53     *
  54     * @task edit
  55     */
  56    public function addEdge($src, $type, $dst, array $options = array()) {
  57      foreach ($this->buildEdgeSpecs($src, $type, $dst, $options) as $spec) {
  58        $this->addEdges[] = $spec;
  59      }
  60      return $this;
  61    }
  62  
  63  
  64    /**
  65     * Remove an edge (possibly also removing its inverse). Changes take effect
  66     * when you call @{method:save}. If an edge does not exist, the removal
  67     * will be ignored. Edges are added after edges are removed, so the effect of
  68     * a remove plus an add is to overwrite.
  69     *
  70     * @param phid  Source object PHID.
  71     * @param const Edge type constant.
  72     * @param phid  Destination object PHID.
  73     * @return this
  74     *
  75     * @task edit
  76     */
  77    public function removeEdge($src, $type, $dst) {
  78      foreach ($this->buildEdgeSpecs($src, $type, $dst) as $spec) {
  79        $this->remEdges[] = $spec;
  80      }
  81      return $this;
  82    }
  83  
  84  
  85    /**
  86     * Apply edge additions and removals queued by @{method:addEdge} and
  87     * @{method:removeEdge}. Note that transactions are opened, all additions and
  88     * removals are executed, and then transactions are saved. Thus, in some cases
  89     * it may be slightly more efficient to perform multiple edit operations
  90     * (e.g., adds followed by removals) if their outcomes are not dependent,
  91     * since transactions will not be held open as long.
  92     *
  93     * @return this
  94     * @task edit
  95     */
  96    public function save() {
  97  
  98      $cycle_types = $this->getPreventCyclesEdgeTypes();
  99  
 100      $locks = array();
 101      $caught = null;
 102      try {
 103  
 104        // NOTE: We write edge data first, before doing any transactions, since
 105        // it's OK if we just leave it hanging out in space unattached to
 106        // anything.
 107        $this->writeEdgeData();
 108  
 109        // If we're going to perform cycle detection, lock the edge type before
 110        // doing edits.
 111        if ($cycle_types) {
 112          $src_phids = ipull($this->addEdges, 'src');
 113          foreach ($cycle_types as $cycle_type) {
 114            $key = 'edge.cycle:'.$cycle_type;
 115            $locks[] = PhabricatorGlobalLock::newLock($key)->lock(15);
 116          }
 117        }
 118  
 119        static $id = 0;
 120        $id++;
 121  
 122        // NOTE: Removes first, then adds, so that "remove + add" is a useful
 123        // operation meaning "overwrite".
 124  
 125        $this->executeRemoves();
 126        $this->executeAdds();
 127  
 128        foreach ($cycle_types as $cycle_type) {
 129          $this->detectCycles($src_phids, $cycle_type);
 130        }
 131  
 132        $this->saveTransactions();
 133      } catch (Exception $ex) {
 134        $caught = $ex;
 135      }
 136  
 137  
 138      if ($caught) {
 139        $this->killTransactions();
 140      }
 141  
 142      foreach ($locks as $lock) {
 143        $lock->unlock();
 144      }
 145  
 146      if ($caught) {
 147        throw $caught;
 148      }
 149    }
 150  
 151  
 152  /* -(  Internals  )---------------------------------------------------------- */
 153  
 154  
 155    /**
 156     * Build the specification for an edge operation, and possibly build its
 157     * inverse as well.
 158     *
 159     * @task internal
 160     */
 161    private function buildEdgeSpecs($src, $type, $dst, array $options = array()) {
 162      $data = array();
 163      if (!empty($options['data'])) {
 164        $data['data'] = $options['data'];
 165      }
 166  
 167      $src_type = phid_get_type($src);
 168      $dst_type = phid_get_type($dst);
 169  
 170      $specs = array();
 171      $specs[] = array(
 172        'src'       => $src,
 173        'src_type'  => $src_type,
 174        'dst'       => $dst,
 175        'dst_type'  => $dst_type,
 176        'type'      => $type,
 177        'data'      => $data,
 178      );
 179  
 180      $type_obj = PhabricatorEdgeType::getByConstant($type);
 181      $inverse = $type_obj->getInverseEdgeConstant();
 182      if ($inverse !== null) {
 183  
 184        // If `inverse_data` is set, overwrite the edge data. Normally, just
 185        // write the same data to the inverse edge.
 186        if (array_key_exists('inverse_data', $options)) {
 187          $data['data'] = $options['inverse_data'];
 188        }
 189  
 190        $specs[] = array(
 191          'src'       => $dst,
 192          'src_type'  => $dst_type,
 193          'dst'       => $src,
 194          'dst_type'  => $src_type,
 195          'type'      => $inverse,
 196          'data'      => $data,
 197        );
 198      }
 199  
 200      return $specs;
 201    }
 202  
 203  
 204    /**
 205     * Write edge data.
 206     *
 207     * @task internal
 208     */
 209    private function writeEdgeData() {
 210      $adds = $this->addEdges;
 211  
 212      $writes = array();
 213      foreach ($adds as $key => $edge) {
 214        if ($edge['data']) {
 215          $writes[] = array($key, $edge['src_type'], json_encode($edge['data']));
 216        }
 217      }
 218  
 219      foreach ($writes as $write) {
 220        list($key, $src_type, $data) = $write;
 221        $conn_w = PhabricatorEdgeConfig::establishConnection($src_type, 'w');
 222        queryfx(
 223          $conn_w,
 224          'INSERT INTO %T (data) VALUES (%s)',
 225          PhabricatorEdgeConfig::TABLE_NAME_EDGEDATA,
 226          $data);
 227        $this->addEdges[$key]['data_id'] = $conn_w->getInsertID();
 228      }
 229    }
 230  
 231  
 232    /**
 233     * Add queued edges.
 234     *
 235     * @task internal
 236     */
 237    private function executeAdds() {
 238      $adds = $this->addEdges;
 239      $adds = igroup($adds, 'src_type');
 240  
 241      // Assign stable sequence numbers to each edge, so we have a consistent
 242      // ordering across edges by source and type.
 243      foreach ($adds as $src_type => $edges) {
 244        $edges_by_src = igroup($edges, 'src');
 245        foreach ($edges_by_src as $src => $src_edges) {
 246          $seq = 0;
 247          foreach ($src_edges as $key => $edge) {
 248            $src_edges[$key]['seq'] = $seq++;
 249            $src_edges[$key]['dateCreated'] = time();
 250          }
 251          $edges_by_src[$src] = $src_edges;
 252        }
 253        $adds[$src_type] = array_mergev($edges_by_src);
 254      }
 255  
 256      $inserts = array();
 257      foreach ($adds as $src_type => $edges) {
 258        $conn_w = PhabricatorEdgeConfig::establishConnection($src_type, 'w');
 259        $sql = array();
 260        foreach ($edges as $edge) {
 261          $sql[] = qsprintf(
 262            $conn_w,
 263            '(%s, %d, %s, %d, %d, %nd)',
 264            $edge['src'],
 265            $edge['type'],
 266            $edge['dst'],
 267            $edge['dateCreated'],
 268            $edge['seq'],
 269            idx($edge, 'data_id'));
 270        }
 271        $inserts[] = array($conn_w, $sql);
 272      }
 273  
 274      foreach ($inserts as $insert) {
 275        list($conn_w, $sql) = $insert;
 276        $conn_w->openTransaction();
 277        $this->openTransactions[] = $conn_w;
 278  
 279        foreach (array_chunk($sql, 256) as $chunk) {
 280          queryfx(
 281            $conn_w,
 282            'INSERT INTO %T (src, type, dst, dateCreated, seq, dataID)
 283              VALUES %Q ON DUPLICATE KEY UPDATE dataID = VALUES(dataID)',
 284            PhabricatorEdgeConfig::TABLE_NAME_EDGE,
 285            implode(', ', $chunk));
 286        }
 287      }
 288    }
 289  
 290  
 291    /**
 292     * Remove queued edges.
 293     *
 294     * @task internal
 295     */
 296    private function executeRemoves() {
 297      $rems = $this->remEdges;
 298      $rems = igroup($rems, 'src_type');
 299  
 300      $deletes = array();
 301      foreach ($rems as $src_type => $edges) {
 302        $conn_w = PhabricatorEdgeConfig::establishConnection($src_type, 'w');
 303        $sql = array();
 304        foreach ($edges as $edge) {
 305          $sql[] = qsprintf(
 306            $conn_w,
 307            '(src = %s AND type = %d AND dst = %s)',
 308            $edge['src'],
 309            $edge['type'],
 310            $edge['dst']);
 311        }
 312        $deletes[] = array($conn_w, $sql);
 313      }
 314  
 315      foreach ($deletes as $delete) {
 316        list($conn_w, $sql) = $delete;
 317  
 318        $conn_w->openTransaction();
 319        $this->openTransactions[] = $conn_w;
 320  
 321        foreach (array_chunk($sql, 256) as $chunk) {
 322          queryfx(
 323            $conn_w,
 324            'DELETE FROM %T WHERE (%Q)',
 325            PhabricatorEdgeConfig::TABLE_NAME_EDGE,
 326            implode(' OR ', $chunk));
 327        }
 328      }
 329    }
 330  
 331  
 332    /**
 333     * Save open transactions.
 334     *
 335     * @task internal
 336     */
 337    private function saveTransactions() {
 338      foreach ($this->openTransactions as $key => $conn_w) {
 339        $conn_w->saveTransaction();
 340        unset($this->openTransactions[$key]);
 341      }
 342    }
 343  
 344    private function killTransactions() {
 345      foreach ($this->openTransactions as $key => $conn_w) {
 346        $conn_w->killTransaction();
 347        unset($this->openTransactions[$key]);
 348      }
 349    }
 350  
 351  
 352  /* -(  Cycle Prevention  )--------------------------------------------------- */
 353  
 354  
 355    /**
 356     * Get a list of all edge types which are being added, and which we should
 357     * prevent cycles on.
 358     *
 359     * @return list<const> List of edge types which should have cycles prevented.
 360     * @task cycle
 361     */
 362    private function getPreventCyclesEdgeTypes() {
 363      $edge_types = array();
 364      foreach ($this->addEdges as $edge) {
 365        $edge_types[$edge['type']] = true;
 366      }
 367      foreach ($edge_types as $type => $ignored) {
 368        $type_obj = PhabricatorEdgeType::getByConstant($type);
 369        if (!$type_obj->shouldPreventCycles()) {
 370          unset($edge_types[$type]);
 371        }
 372      }
 373      return array_keys($edge_types);
 374    }
 375  
 376  
 377    /**
 378     * Detect graph cycles of a given edge type. If the edit introduces a cycle,
 379     * a @{class:PhabricatorEdgeCycleException} is thrown with details.
 380     *
 381     * @return void
 382     * @task cycle
 383     */
 384    private function detectCycles(array $phids, $edge_type) {
 385      // For simplicity, we just seed the graph with the affected nodes rather
 386      // than seeding it with their edges. To do this, we just add synthetic
 387      // edges from an imaginary '<seed>' node to the known edges.
 388  
 389  
 390      $graph = id(new PhabricatorEdgeGraph())
 391        ->setEdgeType($edge_type)
 392        ->addNodes(
 393          array(
 394            '<seed>' => $phids,
 395          ))
 396        ->loadGraph();
 397  
 398      foreach ($phids as $phid) {
 399        $cycle = $graph->detectCycles($phid);
 400        if ($cycle) {
 401          throw new PhabricatorEdgeCycleException($edge_type, $cycle);
 402        }
 403      }
 404    }
 405  
 406  
 407  }