IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Ignore:
Timestamp:
May 3, 2010, 8:45:22 AM (16 years ago)
Author:
eugene
Message:

updates from trunk

Location:
branches/simmosaic_branches
Files:
2 edited

Legend:

Unmodified
Added
Removed

  • branches/simmosaic_branches

  • branches/simmosaic_branches/Nebulous-Server/lib/Nebulous/Server.pm

    r24637 r27839  
    3636use constant TRANS_RETRY_WAIT   => 1;
    3737
     38# This is the umask hack.
     39umask(0002);
     40
     41# This determines how many entries from the list of volumes sorted by free space are randomized.
     42my $topfew_count = 15;
     43my $max_used_space = 0.98;
    3844# transaction restart/retry regex
    3945my $trans_regex = qr/Deadlock Found|Lock wait timeout exceeded|try restarting transaction|Can't connect to MySQL server/i;
    … …  
    696702    $log->debug("entered - @_");
    697703
    698     my ($key, $vol_name) = validate_pos(@_,
     704    my ($key, $dest_vol_name) = validate_pos(@_,
    699705        {
    700706            type        => SCALAR,
    … …  
    726732    # instance on it.  If all avilable volume already have an instance on them
    727733    # then we should throw an error
    728 
    729     # vol_name overrides the key implied volume
    730     eval {
    731         $key = parse_neb_key($key, $vol_name);
     734    # volume names implied as part of the key are *IGNORED* as the source and
     735    # *SHOULD NOT* be used as the destination either
     736
     737    eval {
     738        $key = parse_neb_key($key);
    732739    };
    733740    $log->logdie("$@") if $@;
    734     $vol_name = $key->volume;
    735741
    736742    my $db  = $self->db($key);
    737743
    738     if (defined $vol_name
     744    # puke if the source volume is bogus, we may want to actually use this as
     745    # the instance to be copied later
     746    if (defined $key->volume
    739747        and not $self->_is_valid_volume_name($key, $key->volume)) {
    740748        unless ($key->hard_volume) {
    741             $log->warn( "$vol_name is not a known volume name" );
    742             $vol_name = undef;
     749            $log->warn($key->volume . " not a known volume name");
    743750        } else {
    744            $log->logdie("$vol_name is not a valid volume name");
    745         }
     751           $log->logdie("$key is not a valid volume name");
     752        }
     753    }
     754    # puke if the source volume is bogus, we may want to actually use this as
     755    # the instance to be copied later
     756    if (defined $dest_vol_name
     757        and not $self->_is_valid_volume_name($key, $dest_vol_name)) {
     758           $log->logdie($key->volume . " is not a valid volume name");
    746759    }
    747760       
    748761    my ($vol_id, $vol_host, $vol_path, $vol_xattr);
    749     if (defined $vol_name) {
     762    if (defined $dest_vol_name) {
    750763        ($vol_id, $vol_host, $vol_path, $vol_xattr)
    751             = $self->_get_storage_volume($key, $vol_name);
     764            = $self->_get_storage_volume($key, $dest_vol_name);
    752765    } else {
    753766        ($vol_id, $vol_host, $vol_path, $vol_xattr)
    … …  
    810823            $db->rollback;
    811824            # handle soft volumes
    812             if (defined $vol_name and not defined $key->hard_volume) {
     825            if (defined $dest_vol_name and not defined $key->hard_volume) {
    813826                $log->debug("retrying with 'any' volume");
    814827                return $self->replicate_object($key->path, 'any');
    … …  
    844857
    845858
    846 sub lock_object
    847 {
    848     my $self = shift;
    849 
    850     my $log = $self->log;
    851     $log->debug( "entered - @_" );
    852 
    853     my ( $key, $type ) = validate_pos( @_,
     859sub prune_object
     860{
     861    my $self = shift;
     862
     863    my $log = $self->log;
     864    $log->debug("entered - @_");
     865
     866    my ($key) = validate_pos(@_,
    854867        {
    855868            type        => SCALAR,
    856869            callbacks   => {
    857                 'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
    858             },
    859         },
    860         {
    861             type        => SCALAR,
    862             callbacks   => {
    863                 'is read or write' => sub { $_[0] =~ /^(?:read|write)$/ },
     870                'is valid object key'
     871                    => sub { $self->_is_valid_object_key($_[0]) },
    864872            },
    865873        },
    … …  
    868876    my $sql = $self->sql;
    869877
    870     # ignore volume
    871878    eval {
    872879        $key = parse_neb_key($key);
    … …  
    876883    my $db  = $self->db($key);
    877884
    878     my $so_id;
    879     my $read_lock;
    880     my $write_lock;
    881 
     885    my $rows_removed = 0;
    882886TRANS: while (1) {
    883887        eval {
     888            # remove key from cache
     889            $self->cache->delete($key->path) if defined $self->cache;
     890
     891            my $so_id;
    884892            {
    885                 # this will set update locks
    886                 my $query = $db->prepare_cached( $sql->get_object_locks );
    887                 my $rows = $query->execute( $key->path );
    888                 unless ( $rows == 1 ) {
    889                     $query->finish;
    890                     die( "storage object does not exist" );
    891                 }
    892 
    893                 my $row = $query->fetchrow_hashref;
     893                my $query = $db->prepare_cached( $sql->find_object_by_ext_id );
     894                $query->execute( $key->path );
     895                $so_id = $query->fetchrow_hashref->{'so_id'};
    894896                $query->finish;
    895 
    896                 $so_id      = $row->{ 'so_id' };
    897                 $read_lock  = $row->{ 'read_lock' };
    898                 $write_lock = $row->{ 'write_lock' };
    899             }
    900 
    901             if ($type eq 'write') {
    902                 # can't set a write lock twice and
    903                 # can't set a write lock if there are read locks
    904                 if ($write_lock) {
    905                     die("can not write lock twice -- retry");
    906                 }
    907                
    908                 if ($read_lock > 0) {
    909                     die("can not write lock after read lock -- retry");
    910                 }
    911 
    912                 {
    913                     my $query = $db->prepare_cached( $sql->set_write_lock );
    914                     my $rows = $query->execute($key->path);
    915                
    916                     # if we affected more then one row something very bad has happened.
    917                     unless ($rows == 1) {
    918                         die("affected row count is $rows instead of 1");
    919                     }
    920 
    921                 }
    922             } elsif ($type eq 'read') {
    923                 # can't set a read lock if there's a write lock
    924                 if ($write_lock) {
    925                     die("can not read lock after write lock -- retry");
    926                 }
    927 
    928                 {
    929                     my $query = $db->prepare_cached( $sql->increment_read_lock );
    930                     my $rows = $query->execute($key->path);
    931                
    932                     # if we affected more then one row something very bad has happened.
    933                     unless ($rows == 1) {
    934                         die("affected row count is $rows instead of 1");
    935                     }
    936                 }
    937             }
    938 
     897            }
     898
     899            # record the path of the innaccesible files for deferred
     900            # deletion
     901            my $rows_copied;
     902            {
     903                my $query = $db->prepare_cached( $sql->copy_dead_instances_to_deleted );
     904                $rows_copied = $query->execute( $so_id );
     905            }
     906
     907            # check to see if there is anything to be done
     908            unless ($rows_copied > 0) {
     909                $db->rollback;
     910                return;
     911            }
     912
     913            # In MySQL you can't select from a table you're deleting rows from so
     914            # we first have to get a list of instances to be removed, and then
     915            # remove them.
     916            my $rows_found;
     917            {
     918                my $query = $db->prepare_cached( $sql->find_dead_instances_by_so_id );
     919                $rows_found = $query->execute( $so_id );
     920                my $i = 0;
     921                while (my $row = $query->fetchrow_hashref) {
     922                    # remove dead instances
     923#                   $log->warn("copied: $rows_copied found: $rows_found removed: $rows_removed");
     924                    my $query = $db->prepare_cached( $sql->delete_instance_by_ins_id);
     925                    $rows_removed += $query->execute( $row->{ins_id} );
     926                }
     927                $query->finish;
     928            }
     929#           $log->warn("copied: $rows_copied found: $rows_found removed: $rows_removed");
     930            # sanity check
     931            die("instances inaccessible ($rows_copied) != instances removed ($rows_removed)")
     932                unless $rows_copied == $rows_removed;
     933           
    939934            $db->commit;
    940935            $log->debug("commit");
    … …  
    955950    $log->debug("leaving");
    956951
    957     return 1;
    958 }
    959 
    960 
    961 sub unlock_object
     952    return $rows_removed;
     953}
     954
     955
     956sub lock_object
    962957{
    963958    my $self = shift;
    … …  
    1000995                # this will set update locks
    1001996                my $query = $db->prepare_cached( $sql->get_object_locks );
    1002                 my $rows = $query->execute($key->path);
    1003                 unless ($rows == 1) {
     997                my $rows = $query->execute( $key->path );
     998                unless ( $rows == 1 ) {
    1004999                    $query->finish;
    1005                     die("storage object does not exist");
     1000                    die( "storage object does not exist" );
    10061001                }
    10071002
    … …  
    10151010
    10161011            if ($type eq 'write') {
    1017                 # can't remove a write lock if it doesn't exist
    1018                 if ($read_lock) {
    1019                     die("can not have a write lock under a read lock");
    1020                 }
    1021 
    1022                 unless ($write_lock) {
    1023                     die("can not remove non-existant write lock");
     1012                # can't set a write lock twice and
     1013                # can't set a write lock if there are read locks
     1014                if ($write_lock) {
     1015                    die("can not write lock twice -- retry");
     1016                }
     1017               
     1018                if ($read_lock > 0) {
     1019                    die("can not write lock after read lock -- retry");
    10241020                }
    10251021
    10261022                {
    1027                     my $query = $db->prepare_cached( $sql->delete_write_lock );
     1023                    my $query = $db->prepare_cached( $sql->set_write_lock );
    10281024                    my $rows = $query->execute($key->path);
    10291025               
    … …  
    10321028                        die("affected row count is $rows instead of 1");
    10331029                    }
     1030
    10341031                }
    10351032            } elsif ($type eq 'read') {
    1036                 # can't remove a read lock if there's a write lock and
    1037                 # can't remove a read lock if there aren't any
     1033                # can't set a read lock if there's a write lock
    10381034                if ($write_lock) {
    1039                     die("can not have a read lock under a write lock");
    1040                 }
    1041                    
    1042                 if ($read_lock == 0) {
    1043                     die("can not remove non-existant read lock");
     1035                    die("can not read lock after write lock -- retry");
    10441036                }
    10451037
    10461038                {
    1047                     my $query = $db->prepare_cached( $sql->decrement_read_lock );
     1039                    my $query = $db->prepare_cached( $sql->increment_read_lock );
    10481040                    my $rows = $query->execute($key->path);
    10491041               
    … …  
    10521044                        die("affected row count is $rows instead of 1");
    10531045                    }
    1054 
    1055                 }
    1056             }
     1046                }
     1047            }
     1048
    10571049            $db->commit;
    10581050            $log->debug("commit");
    … …  
    10711063    }
    10721064
    1073     $log->debug( "leaving" );
     1065    $log->debug("leaving");
    10741066
    10751067    return 1;
    … …  
    10771069
    10781070
    1079 sub setxattr_object
    1080 {
    1081     my $self = shift;
    1082 
    1083     my $log = $self->log;
    1084     $log->debug("entered - @_");
    1085 
    1086     my ($key, $name, $value, $flags) = validate_pos(@_,
     1071sub unlock_object
     1072{
     1073    my $self = shift;
     1074
     1075    my $log = $self->log;
     1076    $log->debug( "entered - @_" );
     1077
     1078    my ( $key, $type ) = validate_pos( @_,
    10871079        {
    10881080            type        => SCALAR,
    … …  
    10941086            type        => SCALAR,
    10951087            callbacks   => {
    1096                 'xattr is in user. namespace'
    1097                     => sub { ($_[0]) =~ qr/^user\./ },
    1098             },
    1099         },
    1100         {
    1101             type        => SCALAR,
    1102         },
    1103         {
    1104             type        => SCALAR,
    1105             callbacks   => {
    1106                 'is read or write' => sub { $_[0] =~ /^(?:create|replace)$/i },
     1088                'is read or write' => sub { $_[0] =~ /^(?:read|write)$/ },
    11071089            },
    11081090        },
    … …  
    11181100
    11191101    my $db  = $self->db($key);
     1102
     1103    my $so_id;
     1104    my $read_lock;
     1105    my $write_lock;
    11201106
    11211107TRANS: while (1) {
    11221108        eval {
    1123             my $query;
    1124 
    1125             if ($flags eq 'create') {
    1126                 $query = $db->prepare_cached( $sql->new_object_xattr );
    1127             } else {
    1128                 # replace
    1129                 $query = $db->prepare_cached( $sql->replace_object_xattr );
    1130             }
    1131 
    1132             # name, value, ext_id
    1133             my $rows = $query->execute($name, $value, $key->path);
    1134             $query->finish;
    1135 
    1136             # if we affected more then one row something very bad has happened.
    1137             if ($flags eq 'create') {
     1109            {
     1110                # this will set update locks
     1111                my $query = $db->prepare_cached( $sql->get_object_locks );
     1112                my $rows = $query->execute($key->path);
    11381113                unless ($rows == 1) {
    1139                     die( "affected row count is $rows instead of 1" );
    1140                 }
    1141             } else {
    1142                 # replace_object_xattr can effect either 1 or 2 rows.  2 rows in
    1143                 # the case of a replace and 1 if the xattr didn't already exist.
    1144                 unless ($rows == 1 or $rows == 2) {
    1145                     die( "affected row count is $rows instead of 2" );
    1146                 }
    1147             }
    1148 
     1114                    $query->finish;
     1115                    die("storage object does not exist");
     1116                }
     1117
     1118                my $row = $query->fetchrow_hashref;
     1119                $query->finish;
     1120
     1121                $so_id      = $row->{ 'so_id' };
     1122                $read_lock  = $row->{ 'read_lock' };
     1123                $write_lock = $row->{ 'write_lock' };
     1124            }
     1125
     1126            if ($type eq 'write') {
     1127                # can't remove a write lock if it doesn't exist
     1128                if ($read_lock) {
     1129                    die("can not have a write lock under a read lock");
     1130                }
     1131
     1132                unless ($write_lock) {
     1133                    die("can not remove non-existant write lock");
     1134                }
     1135
     1136                {
     1137                    my $query = $db->prepare_cached( $sql->delete_write_lock );
     1138                    my $rows = $query->execute($key->path);
     1139               
     1140                    # if we affected more then one row something very bad has happened.
     1141                    unless ($rows == 1) {
     1142                        die("affected row count is $rows instead of 1");
     1143                    }
     1144                }
     1145            } elsif ($type eq 'read') {
     1146                # can't remove a read lock if there's a write lock and
     1147                # can't remove a read lock if there aren't any
     1148                if ($write_lock) {
     1149                    die("can not have a read lock under a write lock");
     1150                }
     1151                   
     1152                if ($read_lock == 0) {
     1153                    die("can not remove non-existant read lock");
     1154                }
     1155
     1156                {
     1157                    my $query = $db->prepare_cached( $sql->decrement_read_lock );
     1158                    my $rows = $query->execute($key->path);
     1159               
     1160                    # if we affected more then one row something very bad has happened.
     1161                    unless ($rows == 1) {
     1162                        die("affected row count is $rows instead of 1");
     1163                    }
     1164
     1165                }
     1166            }
    11491167            $db->commit;
    11501168            $log->debug("commit");
    … …  
    11581176                redo TRANS;
    11591177            }
    1160             $log->logdie("database error: $@");
     1178            $log->logdie("error: $@");
    11611179        }
    11621180        last;
    11631181    }
    11641182
    1165     $log->debug("leaving");
     1183    $log->debug( "leaving" );
    11661184
    11671185    return 1;
    … …  
    11691187
    11701188
    1171 sub getxattr_object
     1189sub setxattr_object
    11721190{
    11731191    my $self = shift;
    … …  
    11761194    $log->debug("entered - @_");
    11771195
    1178     my ($key, $name) = validate_pos(@_,
     1196    my ($key, $name, $value, $flags) = validate_pos(@_,
    11791197        {
    11801198            type        => SCALAR,
    … …  
    11901208            },
    11911209        },
    1192     );
    1193 
    1194     my $sql = $self->sql;
    1195 
    1196     # ignore volume
    1197     eval {
    1198         $key = parse_neb_key($key);
    1199     };
    1200     $log->logdie("$@") if $@;
    1201 
    1202     my $db  = $self->db($key);
    1203 
    1204     my $value;
    1205     eval {
    1206         my $query = $db->prepare_cached( $sql->get_object_xattr );
    1207         # ext_id, name
    1208         my $rows = $query->execute($key->path, $name);
    1209 
    1210         # no rows returned means that the xattr does not exist
    1211         if ($rows == 0) {
    1212             $query->finish;
    1213             die( "xattr $key:$name does not exist" );
    1214         }
    1215         # if we go more then one row bad something very bad has happened.
    1216         unless ($rows == 1) {
    1217             $query->finish;
    1218             die( "affected row count is $rows instead of 1" );
    1219         }
    1220 
    1221         my $row = $query->fetchrow_hashref;
    1222         # XXX: DBI bug? ->finish is needed here even though $query is going out
    1223         # of scope
    1224         $query->finish;
    1225         $value = $row->{ 'value' };
    1226     };
    1227     if ($@) {
    1228         if ($@ =~ /user\..*? does not exist/) {
    1229             # do not log xattr does not exist messages
    1230             die $@;
    1231         }
    1232         $log->logdie("database error: $@") if $@;
    1233     }
    1234 
    1235     $log->debug("leaving");
    1236 
    1237     return $value;
    1238 }
    1239 
    1240 
    1241 sub listxattr_object
    1242 {
    1243     my $self = shift;
    1244 
    1245     my $log = $self->log;
    1246     $log->debug("entered - @_");
    1247 
    1248     my ($key) = validate_pos(@_,
     1210        {
     1211            type        => SCALAR,
     1212        },
    12491213        {
    12501214            type        => SCALAR,
    12511215            callbacks   => {
    1252                 'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
    1253             },
    1254         },
    1255     );
    1256 
    1257     my $sql = $self->sql;
    1258 
    1259     # ignore volume
    1260     eval {
    1261         $key = parse_neb_key($key);
    1262     };
    1263     $log->logdie("$@") if $@;
    1264 
    1265     my $db  = $self->db($key);
    1266 
    1267     my @xattrs;
    1268     eval {
    1269         my $query = $db->prepare_cached( $sql->list_object_xattr );
    1270         # ext_id
    1271         my $rows = $query->execute($key->path);
    1272 
    1273         while (my $row = $query->fetchrow_hashref) {
    1274             push @xattrs, $row->{ 'name' };
    1275         }
    1276     };
    1277     $log->logdie("database error: $@") if $@;
    1278 
    1279     $log->debug("leaving");
    1280 
    1281     return \@xattrs;
    1282 }
    1283 
    1284 
    1285 sub removexattr_object
    1286 {
    1287     my $self = shift;
    1288 
    1289     my $log = $self->log;
    1290     $log->debug("entered - @_");
    1291 
    1292     my ($key, $name) = validate_pos(@_,
    1293         {
    1294             type        => SCALAR,
    1295             callbacks   => {
    1296                 'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
    1297             },
    1298         },
    1299         {
    1300             type        => SCALAR,
    1301             callbacks   => {
    1302                 'xattr is in user. namespace'
    1303                     => sub { ($_[0]) =~ qr/^user\./ },
     1216                'is read or write' => sub { $_[0] =~ /^(?:create|replace)$/i },
    13041217            },
    13051218        },
    … …  
    13181231TRANS: while (1) {
    13191232        eval {
    1320             my $query = $db->prepare_cached( $sql->remove_object_xattr );
    1321             # ext_id, name
    1322             my $rows = $query->execute($key->path, $name);
     1233            my $query;
     1234
     1235            if ($flags eq 'create') {
     1236                $query = $db->prepare_cached( $sql->new_object_xattr );
     1237            } else {
     1238                # replace
     1239                $query = $db->prepare_cached( $sql->replace_object_xattr );
     1240            }
     1241
     1242            # name, value, ext_id
     1243            my $rows = $query->execute($name, $value, $key->path);
    13231244            $query->finish;
    13241245
    1325             # no rows affected means the xattr did not exist
    1326             if ($rows == 0) {
    1327                 die( "xattr $key:$name does not exist" );
    1328             }
    1329 
    13301246            # if we affected more then one row something very bad has happened.
    1331             if ($rows > 1) {
    1332                 die( "affected row count is $rows instead of 1" );
     1247            if ($flags eq 'create') {
     1248                unless ($rows == 1) {
     1249                    die( "affected row count is $rows instead of 1" );
     1250                }
     1251            } else {
     1252                # replace_object_xattr can effect either 1 or 2 rows.  2 rows in
     1253                # the case of a replace and 1 if the xattr didn't already exist.
     1254                unless ($rows == 1 or $rows == 2) {
     1255                    die( "affected row count is $rows instead of 2" );
     1256                }
    13331257            }
    13341258
    … …  
    13471271        }
    13481272        last;
     1273    }
     1274
     1275    $log->debug("leaving");
     1276
     1277    return 1;
     1278}
     1279
     1280
     1281sub getxattr_object
     1282{
     1283    my $self = shift;
     1284
     1285    my $log = $self->log;
     1286    $log->debug("entered - @_");
     1287
     1288    my ($key, $name) = validate_pos(@_,
     1289        {
     1290            type        => SCALAR,
     1291            callbacks   => {
     1292                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
     1293            },
     1294        },
     1295        {
     1296            type        => SCALAR,
     1297            callbacks   => {
     1298                'xattr is in user. namespace'
     1299                    => sub { ($_[0]) =~ qr/^user\./ },
     1300            },
     1301        },
     1302    );
     1303
     1304    my $sql = $self->sql;
     1305
     1306    # ignore volume
     1307    eval {
     1308        $key = parse_neb_key($key);
     1309    };
     1310    $log->logdie("$@") if $@;
     1311
     1312    my $db  = $self->db($key);
     1313
     1314    my $value;
     1315    eval {
     1316        my $query = $db->prepare_cached( $sql->get_object_xattr );
     1317        # ext_id, name
     1318        my $rows = $query->execute($key->path, $name);
     1319
     1320        # no rows returned means that the xattr does not exist
     1321        if ($rows == 0) {
     1322            $query->finish;
     1323            die( "xattr $key:$name does not exist" );
     1324        }
     1325        # if we go more then one row bad something very bad has happened.
     1326        unless ($rows == 1) {
     1327            $query->finish;
     1328            die( "affected row count is $rows instead of 1" );
     1329        }
     1330
     1331        my $row = $query->fetchrow_hashref;
     1332        # XXX: DBI bug? ->finish is needed here even though $query is going out
     1333        # of scope
     1334        $query->finish;
     1335        $value = $row->{ 'value' };
     1336    };
     1337    if ($@) {
     1338        if ($@ =~ /user\..*? does not exist/) {
     1339            # do not log xattr does not exist messages
     1340            die $@;
     1341        }
     1342        $log->logdie("database error: $@") if $@;
     1343    }
     1344
     1345    $log->debug("leaving");
     1346
     1347    return $value;
     1348}
     1349
     1350
     1351sub listxattr_object
     1352{
     1353    my $self = shift;
     1354
     1355    my $log = $self->log;
     1356    $log->debug("entered - @_");
     1357
     1358    my ($key) = validate_pos(@_,
     1359        {
     1360            type        => SCALAR,
     1361            callbacks   => {
     1362                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
     1363            },
     1364        },
     1365    );
     1366
     1367    my $sql = $self->sql;
     1368
     1369    # ignore volume
     1370    eval {
     1371        $key = parse_neb_key($key);
     1372    };
     1373    $log->logdie("$@") if $@;
     1374
     1375    my $db  = $self->db($key);
     1376
     1377    my @xattrs;
     1378    eval {
     1379        my $query = $db->prepare_cached( $sql->list_object_xattr );
     1380        # ext_id
     1381        my $rows = $query->execute($key->path);
     1382
     1383        while (my $row = $query->fetchrow_hashref) {
     1384            push @xattrs, $row->{ 'name' };
     1385        }
     1386    };
     1387    $log->logdie("database error: $@") if $@;
     1388
     1389    $log->debug("leaving");
     1390
     1391    return \@xattrs;
     1392}
     1393
     1394
     1395sub removexattr_object
     1396{
     1397    my $self = shift;
     1398
     1399    my $log = $self->log;
     1400    $log->debug("entered - @_");
     1401
     1402    my ($key, $name) = validate_pos(@_,
     1403        {
     1404            type        => SCALAR,
     1405            callbacks   => {
     1406                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
     1407            },
     1408        },
     1409        {
     1410            type        => SCALAR,
     1411            callbacks   => {
     1412                'xattr is in user. namespace'
     1413                    => sub { ($_[0]) =~ qr/^user\./ },
     1414            },
     1415        },
     1416    );
     1417
     1418    my $sql = $self->sql;
     1419
     1420    # ignore volume
     1421    eval {
     1422        $key = parse_neb_key($key);
     1423    };
     1424    $log->logdie("$@") if $@;
     1425
     1426    my $db  = $self->db($key);
     1427
     1428TRANS: while (1) {
     1429        eval {
     1430            my $query = $db->prepare_cached( $sql->remove_object_xattr );
     1431            # ext_id, name
     1432            my $rows = $query->execute($key->path, $name);
     1433            $query->finish;
     1434
     1435            # no rows affected means the xattr did not exist
     1436            if ($rows == 0) {
     1437                die( "xattr $key:$name does not exist" );
     1438            }
     1439
     1440            # if we affected more then one row something very bad has happened.
     1441            if ($rows > 1) {
     1442                die( "affected row count is $rows instead of 1" );
     1443            }
     1444
     1445            $db->commit;
     1446            $log->debug("commit");
     1447        };
     1448        if ($@) {
     1449            $db->rollback;
     1450            $log->debug("rollback");
     1451            if ($@ =~ $trans_regex) {
     1452                $log->warn("database error, retrying transaction: $@");
     1453                sleep TRANS_RETRY_WAIT;
     1454                redo TRANS;
     1455            }
     1456            $log->logdie("database error: $@");
     1457        }
     1458        last;
    13491459    }
    13501460
    … …  
    14961606    $log->debug("entered - @_");
    14971607
    1498     my ($key, $vol_name) = validate_pos(@_,
     1608    my ($key, $vol_name, $find_invalid) = validate_pos(@_,
    14991609        {
    15001610            type        => SCALAR,
    … …  
    15141624            optional    => 1,
    15151625        },
     1626        {
     1627            # find_invalid
     1628            type        => SCALAR|UNDEF,
     1629            optional    => 1,
     1630        },
    15161631    );
    15171632
    … …  
    15571672        } else {
    15581673            $query = $db->prepare_cached( $sql->get_object_instances );
     1674            my $rows;
     1675            # ext_id, available
     1676            if (defined($find_invalid)) {
     1677                $rows = $query->execute($key->path, 0);
     1678            }
     1679            else {
     1680                $rows = $query->execute($key->path, 1);
     1681            }
     1682            unless ($rows > 0) {
     1683                $query->finish;
     1684                die("no instances available for key: $key");
     1685            }
     1686        }
     1687
     1688        while (my $row = $query->fetchrow_hashref) {
     1689            my $instance = $row->{ 'uri' };
     1690            push @locations, $instance if $instance;
     1691        }
     1692    };
     1693    if ($@) {
     1694        $db->rollback;
     1695        # handle soft volumes
     1696        if (defined $vol_name and not defined $key->hard_volume) {
     1697            $log->debug("retrying with 'any' volume");
     1698            return $self->find_instances($key->path, 'any');
     1699        }
     1700        $log->logdie("database error: $@");
     1701    }
     1702
     1703    # XXX remove this?
     1704    $log->logdie("no instances found") unless (scalar @locations);
     1705
     1706    $log->debug("found: @locations");
     1707
     1708    $log->debug("leaving");
     1709
     1710    return \@locations;
     1711}
     1712
     1713sub find_instances_for_cull
     1714{
     1715    my $self = shift;
     1716
     1717    my $log = $self->log;
     1718    $log->debug("entered - @_");
     1719
     1720    my ($key, $vol_name) = validate_pos(@_,
     1721        {
     1722            type        => SCALAR,
     1723            callbacks   => {
     1724                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
     1725            },
     1726        },
     1727        {
     1728            type        => SCALAR|UNDEF,
     1729#            callbacks   => {
     1730#                # check that the volume name requested is valid
     1731#                'is valid volume name' => sub {
     1732#                    return 1 if not defined $_[0];
     1733#                    $self->_is_valid_volume_name($_[0])
     1734#                },
     1735#            },
     1736            optional    => 1,
     1737        },
     1738    );
     1739
     1740    my $sql = $self->sql;
     1741
     1742#    unless ($key) {
     1743#        $log->warn("key was undefined after validate_pos(), trying again...");
     1744#        return $self->find_instances(@_);
     1745#    }
     1746
     1747    # vol_name overrides the key implied volume
     1748    eval {
     1749        $key = parse_neb_key($key, $vol_name);
     1750    };
     1751    $log->logdie("$@") if $@;
     1752    $vol_name = $key->volume;
     1753
     1754    my $db  = $self->db($key);
     1755
     1756    # the key's volume can't be validiated on input for this method so we have
     1757    # to check it after parsing the key
     1758    if (defined $vol_name
     1759        and not $self->_is_valid_volume_name($key, $key->volume)) {
     1760        if ($key->hard_volume) {
     1761            $log->logdie("$vol_name is not a valid volume name");
     1762        } else {
     1763            $log->warn( "$vol_name is not a known volume name" );
     1764            $vol_name = undef;
     1765        }
     1766    }
     1767
     1768    my @locations;
     1769    eval {
     1770        my $query;
     1771        if ($vol_name) {
     1772            $query = $db->prepare_cached( $sql->get_object_instances_by_vol_name );
     1773            # ext_id, name, available
     1774            my $rows = $query->execute($key->path, $vol_name, 1);
     1775            unless ($rows > 0) {
     1776                $query->finish;
     1777                die("no instances on storage volume or volume is not avaiable for key: $key volume: $vol_name");
     1778            }
     1779        } else {
     1780            $query = $db->prepare_cached( $sql->get_object_instances );
    15591781            # ext_id, available
    15601782            my $rows = $query->execute($key->path, 1);
    … …  
    15641786            }
    15651787        }
    1566 
     1788       
    15671789        while (my $row = $query->fetchrow_hashref) {
    1568             my $instance = $row->{ 'uri' };
    1569             push @locations, $instance if $instance;
     1790#           my $instance_hash  = { uri => $row->{ 'uri' },
     1791#                                  vol_id => $row->{ 'vol_id' },
     1792#                                  cab_id => $row->{ 'cab_id'} };
     1793            my $instance = $row->{ 'uri' };
     1794            push @locations, $row if $instance;
    15701795        }
    15711796    };
    … …  
    16741899                # remove key from cache
    16751900                $self->cache->delete($key->path) if defined $self->cache;
    1676 
    1677                 # record the path of the innaccesible files for deferred
    1678                 # deletion
    1679                 {
    1680                     my $query = $db->prepare_cached( $sql->copy_instances_to_deleted );
    1681                     $query->execute( $so_id );
    1682                 }
    1683                 # remove all instances... not strictly nessicary as the delete
    1684                 # from storage_object should cascade but the fkey was specified
    1685                 # without the cascade
    1686                 {
    1687                     my $query = $db->prepare_cached( $sql->delete_instance_by_so_id );
    1688                     $query->execute( $so_id );
     1901               
     1902                if ($total > $available) {
     1903                    $self->prune_object("$key");
    16891904                }
    16901905               
    … …  
    18922107
    18932108    my ($key, $name, $hard_volume) = @_;
    1894 
     2109   
     2110#    $log->warn("_g_s_v: key:>$key< name:>$name< hard_vol:>$hard_volume<");
    18952111    my $sql = $self->sql;
    18962112    my $db  = $self->db($key);
    … …  
    19032119            $query = $db->prepare_cached( $sql->get_storage_volume_by_name );
    19042120            # %free, name, avaiable, allocate
    1905             $rows = $query->execute(0.95, $name, 1, 1);
    1906             # XXX destinguish between non-existant and unaviable
     2121            $rows = $query->execute($max_used_space, $name, 1, 1);
     2122            # XXX destinguish between non-existant and unavailable
    19072123            unless ($rows > 0) {
    19082124                $query->finish;
    … …  
    19242140            $query = $db->prepare_cached( $sql->get_storage_volume );
    19252141            # %free, avaiable, allocate
    1926             $rows = $query->execute(0.95, 1, 1);
     2142            $rows = $query->execute($max_used_space, 1, 1, $topfew_count);
     2143#           $log->warn("Storage_volume: $rows $topfew_count");
    19272144            # there has to be atleast one storage volume
    19282145            unless ($rows > 0) {
    … …  
    19702187    my $db  = $self->db($key);
    19712188
    1972     my ($vol_id, $vol_host, $vol_path, $xattr);
     2189    my ($vol_id, $vol_host, $vol_path, $xattr, $forbidden_cabinet);
    19732190    eval {
    19742191        my $rows;
    1975         my $query = $db->prepare_cached( $sql->get_replication_volume_for_ext_id );
     2192       
     2193        my $query = $db->prepare_cached( $sql->get_cabinets_for_ext_id );
     2194        $rows = $query->execute($key->path);
     2195        unless ($rows > 0) {
     2196            $query->finish;
     2197            die("Requested key $key does not exist");
     2198        }
     2199        if ($rows == 1) {
     2200            ($forbidden_cabinet) = $query->fetchrow_array;
     2201            unless (defined($forbidden_cabinet)) {
     2202                $forbidden_cabinet = 0;
     2203            }
     2204            $query->finish;
     2205        }
     2206        else {
     2207            $forbidden_cabinet = 0;
     2208            $query->finish;
     2209        }
     2210           
     2211
     2212        $query = $db->prepare_cached( $sql->get_replication_volume_for_ext_id );
    19762213        # ext_id, %free, avaiable, allocate
    1977         $rows = $query->execute($key->path, 0.95, 1, 1);
     2214        $rows = $query->execute($key->path, $max_used_space, 1, 1, $forbidden_cabinet, $topfew_count);
    19782215        # XXX destinguish between non-existant and unaviable
    19792216        unless ($rows > 0) {
    … …  
    20832320
    20842321    # handle "any" volume
    2085     if ($vol_name eq 'any') {
     2322    if (($vol_name eq 'any')||($vol_name eq 'any.0')) {
    20862323        $log->debug( "found volume name $vol_name" );
    20872324        $log->debug( "leaving" );
    … …  
    21082345    $log->debug( "leaving" );
    21092346
    2110     return;
     2347    return 0;
    21112348}
    21122349
Note: See TracChangeset for help on using the changeset viewer.