src/applications/repository/storage/PhabricatorRepositoryWorkingCopyVersion.php

   1 <?php
   2
   3 final class PhabricatorRepositoryWorkingCopyVersion
   4   extends PhabricatorRepositoryDAO {
   5
   6   protected $repositoryPHID;
   7   protected $devicePHID;
   8   protected $repositoryVersion;
   9   protected $isWriting;
  10   protected $lockOwner;
  11   protected $writeProperties;
  12
  13   protected function getConfiguration() {
  14     return array(
  15       self::CONFIG_TIMESTAMPS => false,
  16       self::CONFIG_COLUMN_SCHEMA => array(
  17         'repositoryVersion' => 'uint32',
  18         'isWriting' => 'bool',
  19         'writeProperties' => 'text?',
  20         'lockOwner' => 'text255?',
  21       ),
  22       self::CONFIG_KEY_SCHEMA => array(
  23         'key_workingcopy' => array(
  24           'columns' => array('repositoryPHID', 'devicePHID'),
  25           'unique' => true,
  26         ),
  27       ),
  28     ) + parent::getConfiguration();
  29   }
  30
  31   public function getWriteProperty($key, $default = null) {
  32     // The "writeProperties" don't currently get automatically serialized or
  33     // deserialized. Perhaps they should.
  34     try {
  35       $properties = phutil_json_decode($this->writeProperties);
  36       return idx($properties, $key, $default);
  37     } catch (Exception $ex) {
  38       return null;
  39     }
  40   }
  41
  42   public static function loadVersions($repository_phid) {
  43     $version = new self();
  44     $conn_w = $version->establishConnection('w');
  45     $table = $version->getTableName();
  46
  47     // This is a normal read, but force it to come from the master.
  48     $rows = queryfx_all(
  49       $conn_w,
  50       'SELECT * FROM %T WHERE repositoryPHID = %s',
  51       $table,
  52       $repository_phid);
  53
  54     return $version->loadAllFromArray($rows);
  55   }
  56
  57   public static function loadWriter($repository_phid) {
  58     $version = new self();
  59     $conn_w = $version->establishConnection('w');
  60     $table = $version->getTableName();
  61
  62     // We're forcing this read to go to the master.
  63     $row = queryfx_one(
  64       $conn_w,
  65       'SELECT * FROM %T WHERE repositoryPHID = %s AND isWriting = 1
  66         LIMIT 1',
  67       $table,
  68       $repository_phid);
  69
  70     if (!$row) {
  71       return null;
  72     }
  73
  74     return $version->loadFromArray($row);
  75   }
  76
  77
  78   public static function getReadLock($repository_phid, $device_phid) {
  79     $parameters = array(
  80       'repositoryPHID' => $repository_phid,
  81       'devicePHID' => $device_phid,
  82     );
  83
  84     return PhabricatorGlobalLock::newLock('repo.read', $parameters);
  85   }
  86
  87   public static function getWriteLock($repository_phid) {
  88     $parameters = array(
  89       'repositoryPHID' => $repository_phid,
  90     );
  91
  92     return PhabricatorGlobalLock::newLock('repo.write', $parameters);
  93   }
  94
  95
  96   /**
  97    * Before a write, set the "isWriting" flag.
  98    *
  99    * This allows us to detect when we lose a node partway through a write and
 100    * may have committed and acknowledged a write on a node that lost the lock
 101    * partway through the write and is no longer reachable.
 102    *
 103    * In particular, if a node loses its connection to the database the global
 104    * lock is released by default. This is a durable lock which stays locked
 105    * by default.
 106    */
 107   public static function willWrite(
 108     AphrontDatabaseConnection $locked_connection,
 109     $repository_phid,
 110     $device_phid,
 111     array $write_properties,
 112     $lock_owner) {
 113
 114     $version = new self();
 115     $table = $version->getTableName();
 116
 117     queryfx(
 118       $locked_connection,
 119       'INSERT INTO %T
 120         (repositoryPHID, devicePHID, repositoryVersion, isWriting,
 121           writeProperties, lockOwner)
 122         VALUES
 123         (%s, %s, %d, %d, %s, %s)
 124         ON DUPLICATE KEY UPDATE
 125           isWriting = VALUES(isWriting),
 126           writeProperties = VALUES(writeProperties),
 127           lockOwner = VALUES(lockOwner)',
 128       $table,
 129       $repository_phid,
 130       $device_phid,
 131       0,
 132       1,
 133       phutil_json_encode($write_properties),
 134       $lock_owner);
 135   }
 136
 137
 138   /**
 139    * After a write, update the version and release the "isWriting" lock.
 140    */
 141   public static function didWrite(
 142     $repository_phid,
 143     $device_phid,
 144     $old_version,
 145     $new_version,
 146     $lock_owner) {
 147
 148     $version = new self();
 149     $conn_w = $version->establishConnection('w');
 150     $table = $version->getTableName();
 151
 152     queryfx(
 153       $conn_w,
 154       'UPDATE %T SET
 155           repositoryVersion = %d,
 156           isWriting = 0,
 157           lockOwner = NULL
 158         WHERE
 159           repositoryPHID = %s AND
 160           devicePHID = %s AND
 161           repositoryVersion = %d AND
 162           isWriting = 1 AND
 163           lockOwner = %s',
 164       $table,
 165       $new_version,
 166       $repository_phid,
 167       $device_phid,
 168       $old_version,
 169       $lock_owner);
 170   }
 171
 172
 173   /**
 174    * After a fetch, set the local version to the fetched version.
 175    */
 176   public static function updateVersion(
 177     $repository_phid,
 178     $device_phid,
 179     $new_version) {
 180
 181     $version = new self();
 182     $conn_w = $version->establishConnection('w');
 183     $table = $version->getTableName();
 184
 185     queryfx(
 186       $conn_w,
 187       'INSERT INTO %T
 188         (repositoryPHID, devicePHID, repositoryVersion, isWriting)
 189         VALUES
 190         (%s, %s, %d, %d)
 191         ON DUPLICATE KEY UPDATE
 192           repositoryVersion = VALUES(repositoryVersion)',
 193       $table,
 194       $repository_phid,
 195       $device_phid,
 196       $new_version,
 197       0);
 198   }
 199
 200
 201   /**
 202    * Explicitly demote a device.
 203    */
 204   public static function demoteDevice(
 205     $repository_phid,
 206     $device_phid) {
 207
 208     $version = new self();
 209     $conn_w = $version->establishConnection('w');
 210     $table = $version->getTableName();
 211
 212     queryfx(
 213       $conn_w,
 214       'DELETE FROM %T WHERE repositoryPHID = %s AND devicePHID = %s',
 215       $table,
 216       $repository_phid,
 217       $device_phid);
 218   }
 219
 220 }